Effectful

Operations

Syntax

effectful.ops.syntax.defdata(value: Term[T]) → Expr[T][source]

Constructs a Term that is an instance of its semantic type.

Returns:: An instance of T.
Return type:: Expr[T]

This function is the only way to construct a Term from an Operation.

Note

This function is not likely to be called by users of the effectful library, but they may wish to register implementations for additional types.

Example usage:

This is how callable terms are implemented:

@defdata.register(collections.abc.Callable)
class _CallableTerm[**P, T](Term[collections.abc.Callable[P, T]]):
    def __init__(
        self,
        ty: type,
        op: Operation[..., T],
        *args: Expr,
        **kwargs: Expr,
    ):
        self._op = op
        self._args = args
        self._kwargs = kwargs

    @property
    def op(self):
        return self._op

    @property
    def args(self):
        return self._args

    @property
    def kwargs(self):
        return self._kwargs

    @defop
    def __call__(self: collections.abc.Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T:
        ...

When an Operation whose return type is Callable is passed to defdata(), it is reconstructed as a _CallableTerm, which implements the __call__() method.

class effectful.ops.syntax.ConstructorOperation(default: Callable[[Q], V], name: str | None = None)[source]

classmethod define(constructor: type[T] | Callable[[...], T]) → ConstructorOperation[Any, T][source]

Creates a fresh Operation.

Parameters:

t – May be a type, callable, or Operation. If a type, the operation will have no arguments and return the type. If a callable, the operation will have the same signature as the callable, but with no default rule. If an operation, the operation will be a distinct copy of the operation.
name – Optional name for the operation.

Return type:

ConstructorOperation[Any, TypeVar(T)]

Returns:

A fresh operation.

Note

The result of Operation.define() is always fresh (i.e. Operation.define(f) != Operation.define(f)).

Example usage:

Defining an operation:

This example defines an operation that selects one of two integers:

>>> @Operation.define
... def select(x: int, y: int) -> int:
...     return x

The operation can be called like a regular function. By default, select returns the first argument:

>>> select(1, 2)
1

We can change its behavior by installing a select handler:

>>> from effectful.ops.semantics import handler
>>> with handler({select: lambda x, y: y}):
...     print(select(1, 2))
2

Defining an operation with no default rule:

We can use Operation.define() and the NotHandled exception to define an operation with no default rule:
```
>>> @Operation.define
... def add(x: int, y: int) -> int:
...     raise NotHandled
>>> print(str(add(1, 2)))
add(1, 2)
```
When an operation has no default rule, the free rule is used instead, which constructs a term of the operation applied to its arguments. This feature can be used to conveniently define the syntax of a domain-specific language.

Defining free variables:

Passing Operation.define() a type creates a free variable.

>>> from effectful.ops.semantics import evaluate
>>> x = Operation.define(int, name='x')
>>> y = x() + 1

y is free in x, so it is not fully evaluated:

>>> print(str(y))
__add__(x(), 1)

We bind x by installing a handler for it:

>>> with handler({x: lambda: 2}):
...     print(evaluate(y))
3

Note

Because the result of Operation.define() is always fresh, it’s important to be careful with variable identity.

Two operations with the same name that come from different calls to Operation.define are not equal:

>>> x1 = Operation.define(int, name='x')
>>> x2 = Operation.define(int, name='x')
>>> x1 == x2
False

This means that to correctly bind a variable, you must use the same operation object. In this example, scale returns a term with a free variable x:

>>> x = Operation.define(float, name='x')
>>> def scale(a: float) -> float:
...     return x() * a

Binding the variable x as follows does not work:

>>> term = scale(3.0)
>>> fresh_x = Operation.define(float, name='x')
>>> with handler({fresh_x: lambda: 2.0}):
...     print(str(evaluate(term)))
__mul__(x(), 3.0)

Only the original operation object will work:

>>> from effectful.ops.semantics import fvsof
>>> with handler({x: lambda: 2.0}):
...     print(evaluate(term))
6.0

Defining a fresh Operation:

Passing Operation.define() an Operation creates a fresh operation with the same name and signature, but no default rule.

>>> fresh_select = Operation.define(select)
>>> print(str(fresh_select(1, 2)))
select(1, 2)

The new operation is distinct from the original:

>>> with handler({select: lambda x, y: y}):
...     print(select(1, 2), fresh_select(1, 2))
2 select(1, 2)

>>> with handler({fresh_select: lambda x, y: y}):
...     print(select(1, 2), fresh_select(1, 2))
1 2

class effectful.ops.syntax.DataclassConstructorOperation(default: Callable[[Q], V], name: str | None = None)[source]

class effectful.ops.syntax.ObjectInterpretation[source]

A helper superclass for defining an Interpretation of many Operation instances with shared state or behavior.

You can mark specific methods in the definition of an ObjectInterpretation with operations using the implements() decorator. The ObjectInterpretation object itself is an Interpretation (mapping from Operation to Callable)

>>> from effectful.ops.semantics import handler
>>> @defop
... def read_box():
...     pass
...
>>> @defop
... def write_box(new_value):
...     pass
...
>>> class StatefulBox(ObjectInterpretation):
...     def __init__(self, init=None):
...         super().__init__()
...         self.stored = init
...     @implements(read_box)
...     def whatever(self):
...         return self.stored
...     @implements(write_box)
...     def write_box(self, new_value):
...         self.stored = new_value
...
>>> first_box = StatefulBox(init="First Starting Value")
>>> second_box = StatefulBox(init="Second Starting Value")
>>> with handler(first_box):
...     print(read_box())
...     write_box("New Value")
...     print(read_box())
...
First Starting Value
New Value
>>> with handler(second_box):
...     print(read_box())
Second Starting Value
>>> with handler(first_box):
...     print(read_box())
New Value

implementations: dict[Operation[..., T], Callable[[...], V]] = {}

class effectful.ops.syntax.Scoped(ordinal: Set) → None[source]

A special type annotation that indicates the relative scope of a parameter in the signature of an Operation created with defop() .

Scoped makes it easy to describe higher-order Operation s that take other Term s and Operation s as arguments, inspired by a number of recent proposals to view syntactic variables as algebraic effects and environments as effect handlers.

As a result, in effectful many complex higher-order programming constructs, such as lambda-abstraction, let-binding, loops, try-catch exception handling, nondeterminism, capture-avoiding substitution and algebraic effect handling, can be expressed uniformly using defop() as ordinary Operation s and evaluated or transformed using generalized effect handlers that respect the scoping semantics of the operations.

Warning

Scoped instances are typically constructed using indexing syntactic sugar borrowed from generic types like typing.Generic . For example, Scoped[A] desugars to a Scoped instances with ordinal={A}, and Scoped[A | B] desugars to a Scoped instance with ordinal={A, B} .

However, Scoped is not a generic type, and the set of typing.TypeVar s used for the Scoped annotations in a given operation must be disjoint from the set of typing.TypeVar s used for generic types of the parameters.

Example usage:

We illustrate the use of Scoped with a few case studies of classical syntactic variable binding constructs expressed as Operation s.

>>> from typing import Annotated
>>> from effectful.ops.syntax import Scoped, defop
>>> from effectful.ops.semantics import fvsof
>>> x, y = defop(int, name='x'), defop(int, name='y')

For example, we can define a higher-order operation Lambda() that takes an Operation representing a bound syntactic variable and a Term representing the body of an anonymous function, and returns a Term representing a lambda function:

>>> @defop
... def Lambda[S, T, A, B](
...     var: Annotated[Operation[[], S], Scoped[A]],
...     body: Annotated[T, Scoped[A | B]]
... ) -> Annotated[Callable[[S], T], Scoped[B]]:
...     raise NotHandled

The Scoped annotation is used here to indicate that the argument var passed to Lambda() may appear free in body, but not in the resulting function. In other words, it is bound by Lambda():
```
>>> assert x not in fvsof(Lambda(x, x() + 1))
```
However, variables in body other than var still appear free in the result:
```
>>> assert y in fvsof(Lambda(x, x() + y()))
```

Scoped can also be used with variadic arguments and keyword arguments. For example, we can define a generalized LambdaN() that takes a variable number of arguments and keyword arguments:

>>> @defop
... def LambdaN[S, T, A, B](
...     body: Annotated[T, Scoped[A | B]],
...     *args: Annotated[Operation[[], S], Scoped[A]],
...     **kwargs: Annotated[Operation[[], S], Scoped[A]]
... ) -> Annotated[Callable[..., T], Scoped[B]]:
...     raise NotHandled

This is equivalent to the built-in Operation deffn():

>>> assert not {x, y} & fvsof(LambdaN(x() + y(), x, y))

Scoped and defop() can also express more complex scoping semantics. For example, we can define a Let() operation that binds a variable in a Term body to a value that may be another possibly open Term :

>>> @defop
... def Let[S, T, A, B](
...     var: Annotated[Operation[[], S], Scoped[A]],
...     val: Annotated[S, Scoped[B]],
...     body: Annotated[T, Scoped[A | B]]
... ) -> Annotated[T, Scoped[B]]:
...     raise NotHandled

Here the variable var is bound by Let() in body but not in val :

>>> assert x not in fvsof(Let(x, y() + 1, x() + y()))

>>> fvs = fvsof(Let(x, y() + x(), x() + y()))
>>> assert x in fvs and y in fvs

This is reflected in the free variables of subterms of the result:

>>> assert x in fvsof(Let(x, x() + y(), x() + y()).args[1])
>>> assert x not in fvsof(Let(x, y() + 1, x() + y()).args[2])

analyze(bound_sig: BoundArguments) → frozenset[Operation][source]

Computes a set of bound variables given a signature with bound arguments.

The analyze() methods of Scoped annotations that appear on the signature of an Operation are used by defop() to generate implementations of Operation.__fvs_rule__() underlying alpha-renaming in evaluate() and defdata() and free variable sets in fvsof() .

Specifically, the analyze() method of the Scoped annotation of a parameter computes the set of bound variables in that parameter’s value. The Operation.__fvs_rule__() method generated by defop() simply extracts the annotation of each parameter, calls analyze() on the value given for the corresponding parameter in bound_sig , and returns the results.

Parameters:: bound_sig (BoundArguments) – The inspect.Signature of an Operation together with values for all of its arguments.
Return type:: frozenset[Operation]
Returns:: A set of bound variables.

classmethod extract_operations(value, _seen: set[int] | None = None) → frozenset[Operation][source]

Computes the set of Operation s appearing directly in value .

An Operation counts when it appears as a value in a collection, including as the key of a mapping, which is why this cannot be written in terms of flatten() . It does not count when it is applied to arguments, since the resulting Term is a use of the operation rather than a binding occurrence of it.

Parameters:: value – The value to traverse.
Return type:: frozenset[Operation]
Returns:: The operations that could be bound by a parameter given value.

classmethod infer_annotations(sig: Signature) → Signature[source]

Given a inspect.Signature for an Operation for which only some inspect.Parameter s have manual Scoped annotations, computes a new signature with Scoped annotations attached to each parameter, including the return type annotation.

The new annotations are inferred by joining the manual annotations with a fresh root scope. The root scope is the intersection of all Scoped annotations in the resulting inspect.Signature object.

:class`Operation` s in this root scope are free in the result and in all arguments.

Parameters:: sig (Signature) – The signature of the operation.
Return type:: Signature
Returns:: A new signature with inferred Scoped annotations.

ordinal: Set

effectful.ops.syntax.deffn(body: Scoped(ordinal=frozenset({B, A}))], *args: Scoped(ordinal=frozenset({A}))], **kwargs: Scoped(ordinal=frozenset({A}))]) → Scoped(ordinal=frozenset({B}))][source]

An operation that represents a lambda function.

Parameters:

body (TypeVar(T)) – The body of the function.
args (Operation) – Operations representing the positional arguments of the function.
kwargs (Operation) – Operations representing the keyword arguments of the function.

Return type:

Callable[..., TypeVar(T)]

Returns:

A callable term.

deffn() terms are eliminated by the call() operation, which performs beta-reduction.

Example usage:

Here deffn() is used to define a term that represents the function lambda x, y=1: 2 * x + y:

>>> import random
>>> random.seed(0)

>>> x, y = defop(int, name='x'), defop(int, name='y')
>>> term = deffn(2 * x() + y(), x, y=y)
>>> print(str(term))
deffn(...)
>>> term(3, y=4)
10

Note

In general, avoid using deffn() directly. Instead, use trace() to convert a function to a term because it will automatically create the right free variables.

effectful.ops.syntax.defstream(body: Scoped(ordinal=frozenset({A, B}))], streams: Scoped(ordinal=frozenset({B}))]) → Scoped(ordinal=frozenset({A}))][source]

A higher-order operation that represents a for-expression.

Return type:: Iterable[TypeVar(T)]

effectful.ops.syntax.implements(op: Operation[P, V])[source]

Marks a method in an ObjectInterpretation as the implementation of a particular abstract Operation.

When passed an Operation, returns a method decorator which installs the given method as the implementation of the given Operation.

effectful.ops.syntax.iter_(self: Iterable) → Iterator

Return type:: Iterator[TypeVar(T)]

effectful.ops.syntax.next_(self: Iterator) → T

Return type:: TypeVar(T)

effectful.ops.syntax.trace(value: Callable[[P], T]) → Callable[[P], T][source]

Convert a callable to a term by calling it with appropriately typed free variables.

Example usage:

trace() can be passed a function, and it will convert that function to a term by calling it with appropriately typed free variables:

Return type:: Callable[[ParamSpec(P)], TypeVar(T)]

>>> def incr(x: int) -> int:
...     return x + 1
>>> term = trace(incr)

>>> print(str(term))
deffn(__add__(int(), 1), int)

>>> term(2)
3

Semantics

effectful.ops.semantics.as_tuple(*args) → tuple[source]

Return type:: tuple

effectful.ops.semantics.coproduct(intp: Interpretation, intp2: Interpretation) → Interpretation[source]

The coproduct of two interpretations handles any effect that is handled by either. If both interpretations handle an effect, intp2 takes precedence.

Handlers in intp2 that override a handler in intp may call the overridden handler using fwd(). This allows handlers to be written that extend or wrap other handlers.

Example usage:

The message effect produces a welcome message using two helper effects: greeting and name. By handling these helper effects, we can customize the message.

Return type:: Interpretation

>>> message, greeting, name = defop(str), defop(str), defop(str)
>>> i1 = {message: lambda: f"{greeting()} {name()}!", greeting: lambda: "Hi"}
>>> i2 = {name: lambda: "Jack"}

The coproduct of i1 and i2 handles all three effects.

>>> i3 = coproduct(i1, i2)
>>> with handler(i3):
...     print(f'{message()}')
Hi Jack!

We can delegate to an enclosing handler by calling fwd(). Here we override the name handler to format the name differently.

>>> i4 = coproduct(i3, {name: lambda: f'*{fwd()}*'})
>>> with handler(i4):
...     print(f'{message()}')
Hi *Jack*!

Note

coproduct() allows effects to be overridden in a pervasive way, but this is not always desirable. In particular, an interpretation with handlers that call “internal” private effects may be broken if coproducted with an interpretation that handles those effects. It is dangerous to take the coproduct of arbitrary interpretations. For an alternate form of interpretation composition, see product().

effectful.ops.semantics.fvsof(term: Expr) → Set[Operation][source]

Return the free operations in a term.

An operation belongs to fvsof(t) when it appears free in the term t. This excludes operations like apply or collection constructors that are raised during evaluate but do not appear in t. It also excludes operations that are bound by a Scoped operation. However, it is not restricted to the nullary operations in t.

Example usage:

fvsof includes all unbound operations in a term:

Return type:: Set[Operation]

>>> a = defop(int)
>>> @defop
... def f(x: int, y: int) -> int:
...     raise NotHandled
>>> fvs = fvsof(f(a(), 1))
>>> assert fvs >= {f, a}

fvsof accepts the same values as evaluate, including collections:

>>> fvs = fvsof([a(), {'k': f(0, 1)}])
>>> assert fvs >= {f, a}

effectful.ops.semantics.fwd(*args, **kwargs) → Any[source]

Forward execution to the next most enclosing handler.

fwd() should only be called in the context of a handler.

Parameters:

args – Positional arguments.
kwargs – Keyword arguments.

Return type:

Any

If no positional or keyword arguments are provided, fwd() will forward the current arguments to the next handler.

effectful.ops.semantics.handler(intp: Interpretation)[source]: Install an interpretation by taking a coproduct with the current interpretation.

effectful.ops.semantics.typeof(term: Expr) → type[T][source]

Return the type of an expression.

Example usage:

Type signatures are used to infer the types of expressions.

Return type:: type[TypeVar(T)]

>>> @defop
... def cmp(x: int, y: int) -> bool:
...     raise NotHandled
>>> typeof(cmp(1, 2))
<class 'bool'>

Types can be computed in the presence of type variables.

>>> @defop
... def if_then_else[T](x: bool, a: T, b: T) -> T:
...     raise NotHandled
>>> typeof(if_then_else(True, 0, 1))
<class 'int'>

Types

class effectful.ops.types.Annotation[source]

abstractmethod classmethod infer_annotations(sig: Signature) → Signature[source]

Return type:: Signature

class effectful.ops.types.ApplyOperation(default: Callable[[Q], V], name: str | None = None)[source]: An operation that implements application for an Operation subclass.

type effectful.ops.types.Expr = T | Term: An expression is either a value or a term.

class effectful.ops.types.Interpretation(*args, **kwargs)[source]

An interpretation is a mapping from operations to their implementations.

get(**kwds): Helper for @overload to raise when called.

items()[source]

keys()[source]

values()[source]

exception effectful.ops.types.NotHandled[source]: Raised by an operation when the operation should remain unhandled.

class effectful.ops.types.Operation(default: Callable[[Q], V], name: str | None = None)[source]

An abstract class representing an effect that can be implemented by an effect handler.

Note

Do not instantiate Operation directly. Instead, use define() to define operations.

classmethod define(default: Callable[[Q], V], *, name: str | None = None) → Operation[P, T][source]

classmethod define(cls, t: Callable[[P], T], *, name: str | None = None) → Operation[P, T]

classmethod define(cls, t: type[T], **kwargs) → Operation[(), T]

classmethod define(cls, t: Callable[[P], T], **kwargs) → Operation[P, T]

classmethod define(cls, t: staticmethod[P, T], **kwargs)

classmethod define(default: classmethod, **kwargs)

classmethod define(default: singledispatchmethod, **kwargs)

classmethod define(cls, default: _CustomSingleDispatchCallable, **kwargs)

Creates a fresh Operation.

Parameters:

t – May be a type, callable, or Operation. If a type, the operation will have no arguments and return the type. If a callable, the operation will have the same signature as the callable, but with no default rule. If an operation, the operation will be a distinct copy of the operation.
name (str | None) – Optional name for the operation.

Return type:

Operation[ParamSpec(P), TypeVar(T)]

Returns:

A fresh operation.

Note

The result of Operation.define() is always fresh (i.e. Operation.define(f) != Operation.define(f)).

Example usage:

Defining an operation:

This example defines an operation that selects one of two integers:

>>> @Operation.define
... def select(x: int, y: int) -> int:
...     return x

The operation can be called like a regular function. By default, select returns the first argument:

>>> select(1, 2)
1

We can change its behavior by installing a select handler:

>>> from effectful.ops.semantics import handler
>>> with handler({select: lambda x, y: y}):
...     print(select(1, 2))
2

Defining an operation with no default rule:

We can use Operation.define() and the NotHandled exception to define an operation with no default rule:
```
>>> @Operation.define
... def add(x: int, y: int) -> int:
...     raise NotHandled
>>> print(str(add(1, 2)))
add(1, 2)
```
When an operation has no default rule, the free rule is used instead, which constructs a term of the operation applied to its arguments. This feature can be used to conveniently define the syntax of a domain-specific language.

Defining free variables:

Passing Operation.define() a type creates a free variable.

>>> from effectful.ops.semantics import evaluate
>>> x = Operation.define(int, name='x')
>>> y = x() + 1

y is free in x, so it is not fully evaluated:

>>> print(str(y))
__add__(x(), 1)

We bind x by installing a handler for it:

>>> with handler({x: lambda: 2}):
...     print(evaluate(y))
3

Note

Because the result of Operation.define() is always fresh, it’s important to be careful with variable identity.

Two operations with the same name that come from different calls to Operation.define are not equal:

>>> x1 = Operation.define(int, name='x')
>>> x2 = Operation.define(int, name='x')
>>> x1 == x2
False

This means that to correctly bind a variable, you must use the same operation object. In this example, scale returns a term with a free variable x:

>>> x = Operation.define(float, name='x')
>>> def scale(a: float) -> float:
...     return x() * a

Binding the variable x as follows does not work:

>>> term = scale(3.0)
>>> fresh_x = Operation.define(float, name='x')
>>> with handler({fresh_x: lambda: 2.0}):
...     print(str(evaluate(term)))
__mul__(x(), 3.0)

Only the original operation object will work:

>>> from effectful.ops.semantics import fvsof
>>> with handler({x: lambda: 2.0}):
...     print(evaluate(term))
6.0

Defining a fresh Operation:

Passing Operation.define() an Operation creates a fresh operation with the same name and signature, but no default rule.

>>> fresh_select = Operation.define(select)
>>> print(str(fresh_select(1, 2)))
select(1, 2)

The new operation is distinct from the original:

>>> with handler({select: lambda x, y: y}):
...     print(select(1, 2), fresh_select(1, 2))
2 select(1, 2)

>>> with handler({fresh_select: lambda x, y: y}):
...     print(select(1, 2), fresh_select(1, 2))
1 2

class effectful.ops.types.Term[source]

A term in an effectful computation is a is a tree of Operation applied to values.

abstract property args: Sequence[Expr[Any]]: Abstract property for the arguments.

abstract property kwargs: Mapping[str, Expr[Any]]: Abstract property for the keyword arguments.

abstract property op: Operation[..., T]: Abstract property for the operation.

effectful.ops.types.pretty_operation(value: Operation, ctx)[source]: Pretty print raw operations, including those nested in collections.

effectful.ops.types.pretty_term(value: Term, ctx)[source]

Handlers

LLM

Types

LLM-implemented functions via algebraic effects.

effectful.handlers.llm lets you write Python functions whose bodies are implemented by a large language model, and call them like ordinary code.

## Core concepts

`Skill` — a fully type-annotated Python function whose body is raise NotHandled and whose docstring is a [format string](https://docs.python.org/3/library/string.html#format-string-syntax) prompt. Calling a skill (under a provider) formats its arguments into the prompt, invokes the model, and decodes the response to the skill’s declared return type. Define one with the Skill.define decorator.
`Tool` — a normal Python callable exposed to the model. Its signature and docstring become the schema the model sees; the model calls it by name with JSON arguments and receives the encoded result. Tools in a skill’s lexical scope are offered to the model automatically; because scope is ordinary Python scope, an Agent (or an enclosing function) naturally partitions tools and skills into disjoint sets. Define one with Tool.define.
`Agent` — a class mixin giving each instance a persistent message history, so its Skill methods accumulate conversation context across calls. Instance attributes are available in prompts via {self.attr}. Any class defining a Skill method acquires this behavior automatically (see Skill.__set_name__); inheriting Agent explicitly is optional, and remains the spelling static type checkers understand.
`Encodable` — the type-driven JSON bridge used internally to encode Python values into the model’s context and decode the model’s output (structured return values and tool-call arguments) back into typed Python objects.

## Tool calling and structured output

A skill call is a multi-turn loop, and how it runs – how the two messages are assembled, when the model may call a tool, how its answer is decoded – belongs to the handler implementing it: see effectful.handlers.llm.harness.hooks.AgentLoop.

class effectful.handlers.llm.types.Agent(__agent_id__: str | None = None)[source]

Mixin that gives each instance a persistent LLM message history.

Decorate methods with Skill.define. Each instance accumulates messages across calls so the LLM sees prior conversation context.

Subclassing Agent is optional: any class that defines a Skill method in its body acquires this behavior automatically – Skill.__set_name__ copies Agent’s attributes onto the class and registers it as a virtual subclass, so isinstance(obj, Agent) holds for its instances. Explicit inheritance remains supported and is what static type checkers understand (an auto-agentified class’s __agent_id__/__history__ are invisible to them). Skills attached to a class after creation via setattr, classes containing only Tool methods, and fully __slots__-ed classes (whose instances have no __dict__ to hold the history) are not auto-agentified.

Agents compose freely with dataclasses.dataclass and other base classes. Instance attributes are available in skill docstrings via {self.attr}.

Set self.__agent_id__ (a plain attribute, read lazily – see below) to make this instance’s history and declared dataclass fields persist across process restarts when a persistence handler (see effectful.handlers.llm.harness.durability.persistence.SQLitePersister) is installed. Leave it unset (the default) for a normal, transient instance – it still gets a private history, just not backed by any database, and it is never checkpointed even if a persistence handler happens to be active.

Agent itself is deliberately not a dataclass (making it one would make every subclass, even ones with a hand-written __init__, look like a dataclass too – dataclasses.is_dataclass() is inherited – which breaks any such subclass under effectful’s generic dataclass-replace evaluation machinery). Nothing here depends on constructor timing, so there’s no chaining requirement of any kind: __agent_id__ and __persistent__ are derived lazily, on first access, from whatever self.__agent_id__ happens to be at that point – a subclass just needs self.__agent_id__ to end up set to a stable string, however it prefers to do that (a @dataclass field, a custom __init__, or nothing at all, for a transient instance).

Don’t force access to __history__ from within your own __init__ – it’s meant to load lazily, on first real use, not at construction time.

Example:

```python @dataclass class ChatBot(Agent):

bot_name: str

@Skill.define def send(self, user_input: str) -> str:

“””Friendly bot named {self.bot_name}. User writes: {user_input}”””

def main():: chatbot = ChatBot() chatbot.send(“Hi! How are you? I am in France.”) chatbot.send(“Remind me again, where am I?”) # sees prior context

```

## Encapsulation via lexical scope

Since scope is ordinary Python scope, defining agents inside a function partitions their `Skill`s and `Tool`s into disjoint sets:

```python class Chatbot(Agent):

@Skill.define def respond(self, user_query: str) -> str: …

class TravelAdvisor(Agent):

@Skill.define def recommend(self, user_query: str) -> str: … @Tool.define def search_weather(self, city: str) -> str: …

def main():

chatbot, advisor = Chatbot(), TravelAdvisor()

@Skill.define def simulate(chatbot, advisor) -> str:

“””Use {chatbot} and {advisor} to simulate a conversation.””” …

```

chatbot.respond sees only its own methods (plus module-level definitions), not advisor’s; simulate sees chatbot and advisor, but they cannot see simulate. Inlining these definitions into module scope instead would let every skill see every other. Agents that need overlapping toolsets should share tools through a common base class or mixin rather than redefining them.

type effectful.handlers.llm.types.Encodable = Annotated[T, 'encoded'][source]

class effectful.handlers.llm.types.Skill(default: Callable[[P], T], name: str | None = None)[source]

A Skill is a function that is implemented by a large language model.

## Constructing Skills

Apply Skill.define as a decorator to a fully type-annotated function or method whose body is either empty or raise NotHandled. The docstring is a [format string](https://docs.python.org/3/library/string.html#format-string-syntax) prompt: its {…} fields are filled at call time (see effectful.handlers.llm.harness.hooks.AgentLoop, which assembles the messages) and the LLM’s response is decoded to the return type.

Skill.define validates the definition and raises if:

the function has no docstring (every Tool needs one);
a {…} field names something that is neither a parameter nor a name in lexical scope — every field must resolve at call time;
a doctest example (>>>) in the docstring contains an active {…} field: doctests must be constant, since the whole docstring is formatted into the prompt at call time; escape any literal braces as {{ and }}.

See effectful.ops.types.Operation.define for more on Skill.define.

The following skill writes limericks on a given theme:

```python @Skill.define def limerick(theme: str) -> str:

“””Write a limerick on the theme of {theme}. Do not use any tools.”””

```

## Structured output

Skills may return types that are not strings. The output from the LLM is then decoded before being returned to the user.

For example, this skill returns integers:

```python @Skill.define def primes(first_digit: int) -> int:

“””Give a prime number with {first_digit} as the first digit. Do not use any tools.”””

```

Structured generation is used to constrain the LLM to return values that can be decoded without error.

Skills can return complex data structures, such as dataclasses:

```python @dataclass class KnockKnockJoke:

whos_there: str punchline: str

@Skill.define def write_joke(theme: str) -> KnockKnockJoke:

“””Write a knock-knock joke on the theme of {theme}. Do not use any tools.”””

```

Many common Python data types are decodable without additional effort. To register a decoder for a custom type, see effectful.handlers.llm.encoding.type_to_encodable_type.

## Using tools

Instances of Tool in a Skill’s lexical scope may be called by the LLM during completion, and are offered automatically. Scope follows ordinary Python rules: enclosing-function locals, module globals, and — for a method skill — sibling Tool/Skill methods on the same class. A skill cannot call a tool it cannot lexically see, so it should use only tools that are in scope and relevant to the task. Skills are themselves tools, enabling composition into agent workflows.

classmethod define(default: Callable[[Q], V], *args, **kwargs) → Skill[Q, V][source]

Define a skill.

define takes a function and can be used as a decorator. The function’s docstring should be a prompt, which may be templated in the function arguments. The prompt will be provided with any instances of Tool that exist in the lexical context as callable tools.

See effectful.ops.types.Operation.define for more information on the use of Skill.define.

Return type:: Skill[ParamSpec(Q), TypeVar(V)]

effectful.handlers.llm.types.Template: alias of Skill

class effectful.handlers.llm.types.Tool(default: Callable[[P], T], name: str | None = None)[source]

A Tool is a function that may be called by a Skill.

A Tool wraps a normal Python callable; its signature (parameter types and return type) and docstring define the schema the model sees, and the model invokes it by name with JSON arguments.

## Example usage

Skills may call any tool that is in their lexical scope. In the following example, the LLM suggests a vacation destination using the cities and weather tools:

```python @Tool.define def cities() -> list[str]:

“””Return a list of cities that can be passed to weather.””” return [“Chicago”, “New York”, “Barcelona”]

@Tool.define def weather(city: str) -> str:

“””Given a city name, return a description of the weather in that city.””” status = {“Chicago”: “cold”, “New York”: “wet”, “Barcelona”: “sunny”} return status.get(city, “unknown”)

@Skill.define # cities and weather auto-captured from lexical scope def vacation() -> str:

“””Use the cities and weather tools to suggest a city that has good weather.”””

```

Class methods may be used as skills, in which case any other methods decorated with Tool.define will be provided as tools.

classmethod define(default: Callable[[Q], V], *args, **kwargs) → Tool[Q, V][source]

Define a tool.

Binds the result’s type parameters from default (as Skill.define does), so a static checker sees Tool[<params>, <return>] rather than an unbound Tool[Never, Never] – which is what lets a model-written call expression to a tool be type-checked against the tool’s real (possibly generic) signature.

See effectful.ops.types.Operation.define for more information on the use of Tool.define.

Return type:: Tool[ParamSpec(Q), TypeVar(V)]

Harness

Handlers that give the types in effectful.handlers.llm.types their meaning.

The harness() function assembles the standard stack; its constituents are documented in the submodules below and may be recombined or replaced individually.

effectful.handlers.llm.harness.harness(*, num_retries: int = 5, langfuse: bool = False, render: bool = False, dump_system_prompt: str | PathLike[str] | None = None, persist_db: str | PathLike[str] | None = None, eval_provider: Literal['builtin', 'restricted', 'none'] = 'builtin', type_checker: Literal['mypy', 'ty', 'none'] = 'ty', tool_calling: Literal['auto', 'code', 'json'] = 'auto', tool_collection: Literal['none', 'explicit', 'auto'] = 'explicit', check_contracts: bool = True, **provider_config) → Interpretation[source]

Instantiate the standard effectful.handlers.llm handler stack. Install it with handler():

with handler(harness(...)):
    ...

Constructing a harness records the configuration; entering it (as a context manager, decorator, or via the module CLI) installs the handlers and exiting removes them. The handlers, in installation order, are:

AgentLoop, the tool pipeline and LiteLLMConfigurer – the agent loop, the tools it offers from a Skill’s lexical scope, and the model backend it drives. The pipeline is a lexical tool extractor chosen by tool_collection (LexicalToolExtractor for "explicit", ImplicitToolExtractor for "auto", nothing for "none"), plus, above it, the tool caller it feeds (MixedToolCaller for tool_calling="auto", ExpressionToolCaller for "code", none for "json").
FrameworkDocumenter – describe the framework’s concepts in the system prompt.
HistoryBuilder – accumulate the message history of a call.
RichTerminalRenderer – live-render the streaming history (if render).
SystemPromptDumper – dump the system prompt (if dump_system_prompt).
The type_checker and the eval_provider – check and run model-authored Python (each omitted for "none").
StatefulReplSynthesizer and FinalBodySynthesizer – answer a call by running a snippet, and by synthesizing a function and calling it. Both are omitted when eval_provider="none": each advertises a tool (exec_code, write_and_run_body) that only an executor can decode.
PydanticSkillArgValidator – enforce the pre-conditions a caller wrote into a Skill’s parameter annotations (if check_contracts).
TenacityRetryer – retry malformed/failing model output (if num_retries).
SQLitePersister – checkpoint a persisted Agent’s state/history to SQLite after each successful call (if persist_db).
LangfuseTracer – log calls to Langfuse (if langfuse).

Args:

num_retries: Attempts for malformed/failing model output (via: TenacityRetryer, which is left out of the stack altogether when this is 0) and, independently, for transport-level failures (via litellm’s own num_retries, bound into the request).

langfuse: Log LLM calls and metadata to Langfuse. render: Live-render the streaming message history in the terminal. dump_system_prompt: If set, dump the assembled system prompt to this

Markdown file.

persist_db: If set, path to a SQLite database used to checkpoint a: persisted ~effectful.handlers.llm.types.Agent’s (one constructed with an explicit agent_id) state and history via ~effectful.handlers.llm.harness.durability.persistence.SQLitePersister.
eval_provider: Which provider runs model-authored Python:: "builtin" (BuiltinExecutor, the default), "restricted" (RestrictedPythonExecutor), or "none" for no executor – which also takes both synthesizers out of the stack, so nothing is offered that the stack could not then run.
type_checker: Which handler type-checks model-authored Python before it: runs: "ty" (TyTypeChecker, the default), "mypy" (MypyTypeChecker), or "none" to run generated code unchecked.
tool_calling: How the model calls the tools in a Skill’s lexical: scope. "auto" (the default) installs MixedToolCaller, which picks per tool: schema-constrained JSON arguments for every tool a JSON schema can describe faithfully, and the code pathway for the rest (generic, variadic, or unadvertisable signatures). "code" installs ExpressionToolCaller: uniformly, the model writes a Python call expression which is type-checked in the Skill’s scope and evaluated. "json" is the classic JSON-only pathway with no caller at all (polymorphic tools degrade to untyped argument schemas there, and unadvertisable ones are skipped with a warning). "auto" and "code" require an eval provider: combining either with eval_provider="none" raises ValueError rather than silently degrading.
tool_collection: Which tools are collected from a Skill’s lexical: scope, as opposed to how they are called. "explicit" (the default) installs LexicalToolExtractor: the Tool/Skill values in scope, and those held by in-scope Agents. "auto" installs ImplicitToolExtractor instead, which additionally wraps ordinary functions and methods that look deliberately published – public name, docstring, complete annotations (see ImplicitToolExtractor._implicit_tool_candidate) – with no Tool.define decorator; it makes naming conventions load-bearing (prefix orchestration helpers with _ to keep them out of the model’s hands). "none" installs no extractor at all: the model sees only the tools the harness itself injects (exec_code, write_and_run_body), never the surrounding scope’s.
check_contracts: Install PydanticSkillArgValidator, so a Skill’s: arguments are validated against the pydantic metadata its parameter annotations carry. On by default, which makes such an annotation mean the same thing whether a person or a model supplied the argument. Turning it off leaves a direct Python call unchecked; a model-supplied argument is still validated as the tool call is decoded, and metadata on a return annotation is enforced by the decoder either way.

Raises:

ValueError: If tool_calling is "auto" or "code" and: eval_provider is "none".

Return type:: Interpretation

Command-line launcher

A reusable harness for running effectful.handlers.llm example scripts.

The example scripts under docs/source/llm_examples share a fixed stack of handlers – a LiteLLM provider, a Python REPL, retry/decoding logic, and so on – that turns a bare Skill/Agent into something runnable. This module factors that stack into a single object, harness, so the scripts themselves carry none of the boilerplate.

Run as a module it becomes a command-line launcher that wraps an arbitrary script in the same context:

python -m effectful.handlers.llm.harness <path_to_script.py> <harness_flags> <script_flags>

Harness flags are consumed here; other flags pass through to the script unchanged.

effectful.handlers.llm.harness.__main__.main(argv: list[str] | None = None) → None[source]

Return type:: None

Hooks

The operations of the agent loop.

These are the extension points that every other handler in effectful.handlers.llm.harness implements or intercepts.

class effectful.handlers.llm.harness.hooks.AgentLoop[source]

Each turn of this conversation is one request in a loop, and every message you see was assembled by the harness whose capabilities the sections above describe. Here is how those messages are put together.

On each turn you either call tools – results are appended to the conversation and the loop continues – or answer. It is one or the other: a turn that calls a tool is not an answer, so any answer you write alongside a tool call is discarded and you will be asked again. Finish your tool calls first, then answer in a turn of its own. The answer is decoded into the Skill’s declared return type by constrained generation, so a non-str return type (an int, a dataclass, a list of them) comes back to the caller as a real Python value rather than as prose about one. A handler may also mark one of its tools finalizing: calling that tool ends the call, and its return value is the answer (write_and_run_body is the canonical one).

When the return type is str, your message is not summarized or extracted from – the whole of it, verbatim, becomes the value the calling program receives, and is often fed straight into something else. So write the value itself, with no preamble, no sign-off and no commentary about producing it: “Here is the summary you asked for:” is not a preface to the answer, it is part of the answer.

## Calls, turns, and what carries between them

A call is one invocation of a Skill by the program. A turn is one request within it. They are not the same thing, and the conversation you are reading may contain several calls:

Each call opens with exactly one user message. Everything after it – your replies, the tool results answering them – belongs to that call.
When you answer, the call ends and its value is returned to the program, which goes on doing whatever it does with it.
A new user message therefore means the previous call already finished and you are being asked a new question. The one exception says so itself: a user message reporting that your answer could not be decoded is the same call, asked again, and the rest of it is the error to fix.
Only an Agent accumulates history across calls, which is why earlier user messages may be in view at all. For a plain Skill each call starts from an empty conversation.

What survives from one call to the next is exactly what lives outside the conversation: self and the other objects of the Skill’s lexical scope, which belong to the program and are untouched by anything the harness does to the transcript. Per-call machinery does not survive – a REPL session, a submitted implementation, any name you bound while working – so if something you worked out should still be true next time, write it onto self.

## The system message

Assembled once per conversation, in two # halves – the harness the call runs under, then the task itself. The task half is ordered most-constant-first, so the document caches well as the conversation grows:

# | Section heading | Content | Constant over |
- | ————— | ——- | ————- |
| # Harness | A ## subsection per installed handler, each sourced from that handler’s own docstring, describing what this particular stack does — which of them are present varies, so read the headings rather than assuming a fixed set | the handler stack |
| # <name><signature> | The task, introspected from the skill, as the ## subsections below | the conversation |
1 | ## Module <name> | Source of the skill’s module (docstring if source is unavailable) | the module |
2 | ## Agent <cls> (or ## Skill) | Agent docstring, then a ### <name><signature> spec — prompt with {…} holes intact and argument JSON schemas — for every skill sharing the instance’s history (an Agent’s methods, or just this skill) | the instance |
3 | ## Imported modules | Table of in-scope imports (name → module) | the scope |
4 | ## Lexical scope | Table of other in-scope bindings (name → type) | the scope |

Section 1 is contributed entirely by handlers, so a stack that installs none of them omits it; any section that ends up empty is left out of the document entirely.

The whole system message is written once, on the first call of a conversation, and kept as the conversation goes on. So for an Agent calling several of its skills in turn, the # heading above is the first skill called, which need not be the one you are answering now. It is the user message, not this heading, that names the skill of the current turn; section 2.2 specs them all, so the one you need is there either way.

## The user message

The per-call part – written once when the call opens, carrying only what varies between calls; everything constant lives in the system message above. It has two parts, plus whatever the installed handlers add below them:

# | Part | Content |
- | —- | ——- |
1 | Header | <name><signature> — identifies which skill this call is to |
2 | Body | The skill’s docstring with each {…} hole replaced by the encoded value of that argument or in-scope name (non-text values, such as images, as separate content blocks) |

call_agent(skill: Skill[P, T], *args: P, **kwargs: P) → T[source]

The terminal rule: run the completion loop that answers skill.

Assembles the two prompts, then alternates call_assistant and call_tool until a turn produces no tool calls (the model answered) or a tool call reports itself final. Everything else in the harness is a handler layered over the operations this loop invokes, which is why this rule forwards to nothing: it is the bottom of the stack.

Return type:: TypeVar(T)

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function AgentLoop.call_agent>}

type effectful.handlers.llm.harness.hooks.AssistantResult = tuple[ChatCompletionAssistantMessage, Sequence[DecodedToolCall], T | None]

exception effectful.handlers.llm.harness.hooks.DecodingError[source]

Base class for decoding errors that can occur during LLM response processing.

original_error: E

abstractmethod to_feedback_message(*, include_traceback: bool = True) → ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage[source]

Convert the decoding error into a feedback message to be sent back to the LLM.

Return type:: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage

class effectful.handlers.llm.harness.hooks.PromptInjectingInterpretation[source]

Base class for a handler that describes itself to the model.

A subclass’s own docstring becomes a section of the harness half of the system prompt, titled with the class name. That is the whole mechanism: a handler is documented for the model by having a docstring, so what the model is told about a capability and what a reader of the code is told are the same text and cannot drift apart.

A handler with more to say than its docstring – content that has to be computed, like a list built from a constant – overrides this rule, puts its extra sections into harness_prompt, and delegates:

@implements(call_system)
def call_system(self, harness_prompt, agent_prompt):
    return super().call_system(
        PromptSection(
            type="prompt_section",
            title=harness_prompt["title"],
            content=[*harness_prompt["content"], self._extra_section()],
        ),
        agent_prompt,
    )

The base’s own section is appended last of what one handler adds, so extras land immediately ahead of it. Where a handler’s sections sit relative to other handlers’ is decided by installation order alone (see call_system); there is deliberately no per-class knob for it.

call_system(harness_prompt: PromptSection, agent_prompt: PromptSection) → Any[source]

Append this handler’s own class docstring to the harness prompt.

The guard is why a subclass without a docstring stays silent: inspect.getdoc walks the MRO, which is what lets a subclass inherit a parent handler’s description (MixedToolCaller does not restate ExpressionToolCaller’s) – and what would otherwise put the class docstring above, addressed to a reader of this code, into the prompt of every subclass that has none of its own.

Return type:: Any

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

exception effectful.handlers.llm.harness.hooks.ResultDecodingError(original_error: E, raw_message: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage) → None[source]

Error raised when decoding the LLM response result fails.

original_error: E

raw_message: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage

to_feedback_message(*, include_traceback: bool = True) → ChatCompletionUserMessage[source]

Report the failure as a user message.

Return type:: ChatCompletionUserMessage

exception effectful.handlers.llm.harness.hooks.ToolCallDecodingError(original_error: E, raw_message: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage, raw_tool_call: ChatCompletionMessageToolCall) → None[source]

Error raised when decoding a tool call fails.

original_error: E

raw_message: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage

raw_tool_call: ChatCompletionMessageToolCall

to_feedback_message(*, include_traceback: bool = True) → ChatCompletionToolMessage[source]

Convert the decoding error into a feedback message to be sent back to the LLM.

Return type:: ChatCompletionToolMessage

exception effectful.handlers.llm.harness.hooks.ToolCallExecutionError(original_error: E, raw_tool_call: DecodedToolCall) → None[source]

Error raised when a tool execution fails at runtime.

original_error: E

raw_tool_call: DecodedToolCall

to_feedback_message(*, include_traceback: bool = True) → ChatCompletionToolMessage[source]

Convert the decoding error into a feedback message to be sent back to the LLM.

Return type:: ChatCompletionToolMessage

type effectful.handlers.llm.harness.hooks.ToolResult = tuple[ChatCompletionToolMessage, T | ToolCallExecutionError, bool]

effectful.handlers.llm.harness.hooks.call_agent(op: Operation[A, B], *args: A, **kwargs: A) → B

Alias for Skill.__apply__: the operation invoked when a Skill is called.

Handlers install against this to intercept an agent call, alongside the other call_* hooks in this module.

Return type:: TypeVar(B)

effectful.handlers.llm.harness.hooks.call_assistant(messages: Sequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], response_type: type[T], env: Mapping[str, Any], tools: Set[Tool] = frozenset({})) → AssistantResult[source]

Low-level LLM request. Handlers may log/modify requests and delegate via fwd().

This effect is emitted for model request/response rounds so handlers can observe/log requests.

The request is fully determined by the arguments: messages is the conversation sent to the model, so the rule reads no ambient history and a caller (or an intercepting handler) decides exactly what the model sees.

The available tools are passed explicitly as a set; handlers that expose additional tools (synthetic readers, REPL access, synthesis) intercept this operation and union them into tools before forwarding.

Return type:: GenericAlias[TypeVar(T)]

Raises:

ToolCallDecodingError: If a tool call cannot be decoded. The error: includes the raw assistant message for retry handling.
ResultDecodingError: If the result cannot be decoded. The error: includes the raw assistant message for retry handling.

effectful.handlers.llm.harness.hooks.call_system(harness_prompt: PromptSection, agent_prompt: PromptSection) → ChatCompletionSystemMessage[source]

Assemble the system message from the two halves of the system prompt.

agent_prompt describes the task: the caller (AgentLoop.call_agent) introspects it from the Skill being called. harness_prompt describes the machinery the task runs under, and arrives empty: each installed handler with something to say about the harness intercepts this operation and adds the section documenting the capability it provides. This rule only has to put the two together and flatten the result.

Handing the handlers their own argument is what keeps them independent. A handler appends to harness_prompt and forwards; it never looks a section up by title, never has to create one that isn’t there, and cannot disturb the other half of the document. Installation order therefore decides only the order the sections appear in – innermost first, since it intercepts first – and never whether one of them lands.

Return type:: ChatCompletionSystemMessage

effectful.handlers.llm.harness.hooks.call_tool(tool_call: DecodedToolCall) → ToolResult[source]

Implements a roundtrip call to a python function. Input is a json string representing an LLM tool call request parameters. The output is the serialised response to the model.

Returns the appended tool message, the tool’s return value, and whether the call finalizes the Skill – always False here. Finalization is a policy a handler of this operation applies to its own tools, not a property of the tool’s type: see effectful.handlers.llm.harness.synthesis.body.FinalBodySynthesizer, which marks its write_and_run_body call final so that value becomes the Skill’s result and the completion loop stops.

The returned value is a ToolCallExecutionError rather than the tool’s result when a handler captured a failed call (see effectful.handlers.llm.harness.durability.TenacityRetryer); this rule itself raises instead.

Return type:: GenericAlias[TypeVar(T)]

effectful.handlers.llm.harness.hooks.call_user(user_prompt: PromptSection) → ChatCompletionUserMessage[source]

Format a Skill’s prompt applied to arguments into a user message.

user_prompt is wrapped in an enclosing document for the same reason call_system assembles one: _render_prompt_section treats level 0 as the document itself and does not render its title, so a section handed straight to it would lose its heading. Wrapped, user_prompt is a child and its title becomes the message’s # heading.

Return type:: ChatCompletionUserMessage

Low-level LLM request. Handlers may log/modify requests and delegate via fwd().

This effect is emitted for model request/response rounds so handlers can observe/log requests.

Return type:: ModelResponse | CustomStreamWrapper

Serialization

Conversion between Python values and the model’s wire format.

Python values are converted to the content blocks, tool schemas and JSON payloads exchanged with the model, and the model’s output is converted back.

class effectful.handlers.llm.harness.serialization.DecodedToolCall(tool: Tool[..., T], bound_args: BoundArguments, id: ToolCallID, name: str, source: str | None = None) → None[source]

Structured representation of a tool call decoded from an LLM response.

bound_args: BoundArguments

id: ToolCallID

name: str

property result_type: type[T]

source: str | None = None

tool: Tool[..., T]

class effectful.handlers.llm.harness.serialization.EncodedFunction(**data: Any) → None[source]

A function, encoded as a string of its complete Python source.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

code: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class effectful.handlers.llm.harness.serialization.PromptSection[source]

content: Sequence[ChatCompletionTextObject | ChatCompletionImageObject | ChatCompletionAudioObject | ChatCompletionDocumentObject | ChatCompletionVideoObject | ChatCompletionFileObject | PromptSection]

title: str

type: Literal['prompt_section']

type effectful.handlers.llm.harness.serialization.ToolCallID = str

class effectful.handlers.llm.harness.serialization.TypeToPydanticType[source]

Substitute custom types with their Pydantic Annotated equivalents.

Recursively walks a type annotation tree, replacing leaf types that have registered Pydantic annotations (e.g., Image.Image -> PydanticImage) and reconstructing the full generic type.

The result can be passed to pydantic.TypeAdapter() for automatic validation and serialization of nested structures.

evaluate(ty)[source]: Normalize generic types

classmethod register(*args, **kwargs)[source]

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

effectful.handlers.llm.harness.serialization.format_as_content_blocks(template: str, env: Mapping[str, Any]) → list[ChatCompletionTextObject | ChatCompletionImageObject | ChatCompletionAudioObject | ChatCompletionDocumentObject | ChatCompletionVideoObject | ChatCompletionFileObject][source]

Format a template applied to arguments into a list of content blocks. This is similar to str.format() but produces a list of content blocks instead of a single string, so that non-text content is preserved.

Return type:: list[ChatCompletionTextObject | ChatCompletionImageObject | ChatCompletionAudioObject | ChatCompletionDocumentObject | ChatCompletionVideoObject | ChatCompletionFileObject]

effectful.handlers.llm.harness.serialization.to_content_blocks(value: Any) → Sequence[ChatCompletionTextObject | ChatCompletionImageObject | ChatCompletionAudioObject | ChatCompletionDocumentObject | ChatCompletionVideoObject | ChatCompletionFileObject][source]

Convert an encoded JSON-compatible value into a flat list of content blocks.

Walks the value tree, extracting content-block-shaped dicts (identified by their type discriminator) and emitting JSON syntax as text around them.

Top-level strings are emitted bare (for natural template rendering). Inside JSON structures, separators match json.dumps defaults so that the linearization law holds for non-string encoded values: linearize(to_content_blocks(v)) == json.dumps(v).

Return type:: Sequence[ChatCompletionTextObject | ChatCompletionImageObject | ChatCompletionAudioObject | ChatCompletionDocumentObject | ChatCompletionVideoObject | ChatCompletionFileObject]

Provision

Handlers that bind the agent loop to a model backend.

class effectful.handlers.llm.harness.provision.litellm.LiteLLMConfigurer(model='gpt-4o', **config)[source]

Configures the LiteLLM API, and enforces the parts of that configuration that a provider may ignore (see _enforce_tool_choice).

completion(*args, **kwargs)[source]

Inject the provider’s configuration (model and bound litellm kwargs) into the low-level request before delegating, and hold the response to the tool_choice that request carries.

The merge below lets a value already in kwargs stand, so an enclosed configurer’s tool_choice is the one litellm is sent – and, being the one sent, the only one there is anything to enforce about. Reading it back off the merged request rather than off self.config is what keeps those two in agreement.

config: Mapping[str, Any]

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(completion, (model: str, messages: list = [], timeout: float | str | openai.Timeout | None = None, temperature: float | None = None, top_p: float | None = None, n: int | None = None, stream: bool | None = None, stream_options: dict | None = None, stop=None, max_completion_tokens: int | None = None, max_tokens: int | None = None, modalities: list[Literal['text', 'audio']] | None = None, prediction: openai.types.chat.chat_completion_prediction_content_param.ChatCompletionPredictionContentParam | None = None, audio: openai.types.chat.chat_completion_audio_param.ChatCompletionAudioParam | None = None, presence_penalty: float | None = None, frequency_penalty: float | None = None, logit_bias: dict | None = None, user: str | None = None, reasoning_effort: Literal['none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'default'] | None = None, verbosity: Literal['low', 'medium', 'high'] | None = None, response_format: dict | type[pydantic.main.BaseModel] | None = None, seed: int | None = None, tools: list | None = None, tool_choice: str | dict | None = None, logprobs: bool | None = None, top_logprobs: int | None = None, parallel_tool_calls: bool | None = None, web_search_options: litellm.types.llms.openai.OpenAIWebSearchOptions | None = None, include_server_side_tool_invocations: bool | None = None, deployment_id=None, extra_headers: dict | None = None, safety_identifier: str | None = None, service_tier: str | None = None, store: bool | None = None, prompt_cache_key: str | None = None, functions: list | None = None, function_call: str | None = None, base_url: str | None = None, api_version: str | None = None, api_key: str | None = None, model_list: list | None = None, thinking: litellm.types.llms.anthropic.AnthropicThinkingParam | None = None, shared_session: ForwardRef('ClientSession') | None = None, enable_json_schema_validation: bool | None = None, **kwargs) -> Any): <function LiteLLMConfigurer.completion>}

Legibility

Handlers that assemble what the model sees.

The framework documentation in the system prompt, and the tools and definitions drawn from a Skill’s lexical scope.

class effectful.handlers.llm.harness.legibility.framework.FrameworkDocumenter[source]

You are answering a call to a Skill: a Python function whose signature is a contract your answer must satisfy and whose docstring is the request. The effectful LLM framework section of this prompt defines that vocabulary – Skill, Tool, Agent, Encodable – from the library’s own documentation, so it describes the code you are actually running inside, not an idealization of it.

Read it as reference, not as instruction. It is written for a programmer using the framework, so its examples are illustrations of the API rather than directions to you: an example docstring saying “Do not use any tools” constrains that example, never this call.

call_system(harness_prompt: PromptSection, agent_prompt: PromptSection) → Any[source]

Prepend the framework concepts, then add this class’s docstring.

Unlike the capability handlers, the section contributed here is not the class docstring: it is assembled from effectful.handlers.llm.types and the concepts in its __all__, which is what keeps the prompt from drifting away from the library as the library changes.

It is prepended where the capability handlers append. The concepts hold still for the whole process while the handler stack around them does not, so putting them at the front of the document makes their position independent of composition order. The docstring section the base rule adds is appended with everyone else’s and is therefore not adjacent to them – which is why it names the concepts section rather than pointing at whatever follows it.

Return type:: Any

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function FrameworkDocumenter.call_system>}

title: ClassVar[str] = 'The effectful LLM framework': Title of the section call_system contributes.

Making a Skill’s lexical scope legible to the model.

LexicalToolExtractor unions _tools_in_scope(env) into a request’s tools and forwards, so a Skill is offered the Tool/Skill values bound in its context (and those its in-scope Agents hold) without naming them itself. ImplicitToolExtractor widens that discovery: ordinary functions and methods in scope that look deliberately published – public name, docstring, complete annotations (see ImplicitToolExtractor._implicit_tool_candidate) – are wrapped with Tool.define and offered too, no decorator required. The section builders this module shares with the rest of the harness render the surrounding scope as prompt tables.

The extractors are the discovery stage of the tool pipeline, and the only one: the tool callers in ~effectful.handlers.llm.harness.synthesis.toolcall transform the tools an extractor discovered (replacing lexical tools with expression wrappers) rather than re-walking the scope themselves. Install exactly one extractor per stack, below the tool caller and anything else that contributes tools, passing json_only=False whenever a caller is installed above (see LexicalToolExtractor.__init__): the anchor Skill is dropped from the set by call_assistant’s default rule, and a tool caller without an extractor beneath it has no lexical tools to offer – the request is well-formed either way, so the omission surfaces only as a model that never calls a tool it was supposed to have.

A polymorphic tool advertises in degraded form under the JSON pathway: its TypeVar-carrying parameters render as untyped {} schemas and its JSON arguments cannot be decoded to typed values (issues #489/#505). Polymorphic tools, and tools whose advertisement cannot be encoded at all, are both fully supported by the code-generation pathway (~effectful.handlers.llm.harness.synthesis.toolcall.ExpressionToolCaller), which composes above either extractor.

class effectful.handlers.llm.harness.legibility.lexical.ImplicitToolExtractor(predicate: Callable[[LambdaType], bool] | None = None, json_only: bool = True)[source]

The tools you are offered are the ones this Skill can actually reach in the surrounding code: the Tool and Skill values bound in its lexical scope, those held by any Agent in that scope – and, beyond the ones declared as tools, the ordinary functions and methods of that scope that are public, documented, and fully type-annotated, wrapped and offered as tools automatically. Nobody chose them by hand: the set is what the surrounding code has in view, so read it as evidence of what the caller expects this task to need, and read each tool’s own docstring as its contract.

That has a practical consequence: a capability you might expect is missing from the list because it is not in scope here (or is private, undocumented, or unannotated), not because it is forbidden. Do not try to name or invoke a tool that is not offered. If the work seems to require one, do what you can with what is offered and say plainly what was missing.

The Lexical scope and Imported modules tables list the same scope’s non-callable bindings, so the tools and those tables describe one environment together.

json_only says whether this extractor feeds the JSON pathway directly, with no tool caller stacked above it.

When true (the default, and how tool_calling="json" installs it), a discovered tool whose JSON advertisement cannot be encoded is dropped, with a warning, before it can fail the encoding of the whole request. When a caller is installed above (~effectful.handlers.llm.harness.synthesis.toolcall.MixedToolCaller or ExpressionToolCaller), pass json_only=False – the caller replaces exactly those tools with expression wrappers (which always encode), so dropping them here would starve the expression pathway of the tools it exists for.

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function LexicalToolExtractor.call_assistant>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

class effectful.handlers.llm.harness.legibility.lexical.LexicalToolExtractor(json_only: bool = True)[source]

The tools you are offered are the ones this Skill can actually reach: the Tool and Skill values bound in its lexical scope, plus those held by any Agent in that scope. Nobody chose them for you by hand – they are what the surrounding code has in view – so the set is worth reading as evidence of what the caller expects this task to need.

That has a practical consequence: a capability you might expect is missing from the list because it is not in scope here, not because it is forbidden. Do not try to name or invoke a tool that is not offered. If the work seems to require one, do what you can with what is offered and say plainly what was missing.

The Lexical scope and Imported modules tables list the same scope’s non-callable bindings, so the tools and those tables describe one environment together.

json_only says whether this extractor feeds the JSON pathway directly, with no tool caller stacked above it.

When true (the default, and how tool_calling="json" installs it), a discovered tool whose JSON advertisement cannot be encoded is dropped, with a warning, before it can fail the encoding of the whole request. When a caller is installed above (~effectful.handlers.llm.harness.synthesis.toolcall.MixedToolCaller or ExpressionToolCaller), pass json_only=False – the caller replaces exactly those tools with expression wrappers (which always encode), so dropping them here would starve the expression pathway of the tools it exists for.

call_assistant(messages: Sequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], response_type: type, env: Mapping[str, Any], tools: Set[Tool] = frozenset({})) → AssistantResult[source]

Union the in-scope tools into the request.

Discovery: this handler decides which tools the Skill’s scope offers; a tool caller stacked above it may then replace lexical tools with expression wrappers. Under json_only (no caller above), each discovered tool’s advertisement is probed first, and one that cannot be encoded is skipped with a warning – implicitly wrapped tools (see ImplicitToolExtractor) log at debug instead, since a blanket scan of a scope legitimately picks up functions that don’t advertise.

Return type:: TypeAliasType

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function LexicalToolExtractor.call_assistant>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

Execution

Operations and handlers for parsing, compiling and running model-authored Python.

Either with the builtins or under a RestrictedPython policy.

effectful.handlers.llm.harness.execution.hooks.compile(source: str | AST, filename: str, mode: str = 'exec', flags: int = 0, dont_inherit: bool = False, optimize: int = -1) → CodeType[source]

Compile source text or an AST into a Python code object.

Takes builtins.compile’s signature, so it can stand in for the builtin wherever one is called positionally – notably inside doctest’s runner, which run_doctests redirects here. Only mode differs, defaulting to "exec" (the module compile that synthesis does) rather than being required.

Return type:: CodeType

source: The source to compile: an AST (typically produced by parse()) or the: source text of one.

filename: The filename recorded in the resulting code object (CodeType.co_filename), used in tracebacks and by inspect.getsource(). mode: "exec", "eval" or "single", as for builtins.compile. flags, dont_inherit, optimize: as for builtins.compile.

Returns the compiled code object.

effectful.handlers.llm.harness.execution.hooks.eval(bytecode: CodeType, env: dict[str, Any]) → Any[source]

Evaluate a compiled expression code object and return its value.

Return type:: Any

bytecode: A code object compiled in "eval" mode (typically produced by: compile(…, mode=”eval”)).

env: The namespace mapping used during evaluation.

Returns the expression’s value. Binding effects are discarded: unlike exec, env is not updated after evaluation – the only construct that could bind a name from eval-mode code is a scope-escaping walrus, and callers that must not observe one reject it before compiling.

Deliberately (bytecode, env), symmetric with the sibling exec operation, rather than builtins.eval’s (source, globals, locals). Only compile mirrors its builtin, and only because run_doctests rebinds doctest.compile to it positionally; nothing stands this operation in for the builtin, globals=None (“use the caller’s frame”) is meaningless as an effect operation, and accepting str source would collapse the parse -> compile -> eval separation the operations are built on.

effectful.handlers.llm.harness.execution.hooks.exec(bytecode: CodeType, env: dict[str, Any]) → None[source]

Execute a compiled code object.

bytecode: A code object to execute (typically produced by compile()). env: The namespace mapping used during execution.

After exec(bytecode, env) returns, env reflects all top-level binding effects of the executed code (new names and rebindings alike).

Return type:: None

effectful.handlers.llm.harness.execution.hooks.parse(source: str, filename: str) → Module[source]

Parse source text into an AST.

source: The Python source code to parse. filename: The filename recorded in the resulting AST for tracebacks and tooling.

Returns the parsed AST.

Return type:: Module

An UNSAFE eval provider built on the interpreter’s own builtins.

BuiltinExecutor implements the parse/compile/eval/exec operations by calling ast.parse, builtins.compile, builtins.eval and builtins.exec directly, in this process, without any further checks – generated code gets the full authority of the interpreter running the harness. Only use it for testing, or where the code being run is trusted for some other reason; ~effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor is the provider for anything else.

It runs whatever it is given: type checking is a separate handler (~effectful.handlers.llm.harness.validation.mypy.MypyTypeChecker or ~effectful.handlers.llm.harness.validation.ty.TyTypeChecker), installed alongside this one when generated code should be checked before it runs.

class effectful.handlers.llm.harness.execution.builtin.BuiltinExecutor[source]

Code you write runs as ordinary Python in this process, with nothing restricting it. The whole standard library is available, any installed third-party package is importable, and the filesystem, the network and the process itself are all reachable. If an import would work in a normal Python session, it works here.

So write straightforward code and import what you need instead of working around a sandbox that is not there. The corresponding responsibility is yours: the same lack of restriction means a stray open(…, “w”) or a subprocess call really does touch the machine. Do the work the request asks for and nothing else with side effects beyond it.

compile(source: str | AST, filename: str, mode: str = 'exec', flags: int = 0, dont_inherit: bool = False, optimize: int = -1) → CodeType[source]

Compile source – text or AST – with builtins.compile.

A straight pass-through: unlike ~effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor.compile, there is no policy to apply and nothing to rewrite, so the flags the caller passes are the flags CPython sees.

Return type:: CodeType

eval(bytecode: CodeType, env: dict[str, Any]) → Any[source]

Evaluate bytecode for its value, discarding its binding effects.

The evaluation happens in a copy of env, because the operation’s contract is that eval yields a value and changes nothing: a walrus in the expression – or the __builtins__ entry seeded here, which the caller never asked for – must not leak back into the caller’s environment. exec is the operation that does bind.

Return type:: Any

exec(bytecode: CodeType, env: dict[str, Any]) → None[source]

Execute bytecode in env, keeping whatever it binds.

Unlike eval, this runs against env itself, and passes it as both globals and locals so execution is module-style: a top-level def or assignment lands in env and is visible to the next statement executed there, which is what makes a sequence of snippets behave like a session rather than a series of unrelated fragments.

Return type:: None

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, Operation(parse, (source: str, filename: str) -> ast.Module): <function BuiltinExecutor.parse>, Operation(compile, (source: str | ast.AST, filename: str, mode: str = 'exec', flags: int = 0, dont_inherit: bool = False, optimize: int = -1) -> code): <function BuiltinExecutor.compile>, Operation(eval, (bytecode: code, env: dict[str, typing.Any]) -> Any): <function BuiltinExecutor.eval>, Operation(exec, (bytecode: code, env: dict[str, typing.Any]) -> NoneType): <function BuiltinExecutor.exec>}

parse(source: str, filename: str) → Module[source]

Parse source, registering it under filename so it stays readable.

Generated source has no file behind it, so inspect.getsource on a function defined here would otherwise fail. Seeding linecache under the same name the code object will carry (inspect goes from f.__code__.co_filename to linecache.getlines(filename)) makes the synthesized code introspectable like any other – which is what lets a traceback show real lines, and a later turn read back what it wrote.

Return type:: Module

A safer eval provider built on RestrictedPython.

RestrictedPython is not a complete sandbox: it enforces a restricted language subset at compile time and expects the caller to supply a constrained exec environment. RestrictedPythonExecutor supplies that environment – a RestrictedPythonPolicy at compile time, and at run time the guarded accessors that policy’s output calls into (_guarded_getattr, _guarded_import, …) over a builtins namespace with no I/O, no introspection and no way back to compile/eval/exec.

It also cannot shut down the process it runs in. call_tool catches Exception and not BaseException, so that a real Ctrl-C interrupts the host rather than being handed to the model as a failed tool call – which leaves anything a snippet can raise outside Exception uncontainable. So the BaseException subclasses that are not Exception are removed from the builtins namespace (_UNSAFE_BUILTIN_NAMES), and os/signal are out of reach whether named in an import or fetched off an allowed module that imported them itself (_checked_module).

Doctests are executed under the same policy as the code they exercise, so a model cannot smuggle past the sandbox in a docstring.

The sandbox says nothing about types: install ~effectful.handlers.llm.harness.validation.mypy.MypyTypeChecker or ~effectful.handlers.llm.harness.validation.ty.TyTypeChecker alongside this handler to type-check generated code before it is compiled and run.

class effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor(*, policy: type[RestrictingNodeTransformer] | None = None)[source]

Code you write runs in a restricted subset of Python, not the full language. What is unavailable is unavailable by design, and no amount of indirection will reach it, so write within the subset rather than testing its edges – a rejected program costs a turn and tells you only what you already know from here.

The restrictions: no file, network or process access, and no open, input, eval, exec, compile, globals, locals, vars, dir or breakpoint. Imports are limited to the allowlist in the Modules you may import section, and a module not on it stays out of reach however you get to it – naming it in an import, or taking it off an allowed module that imported it. Introspection back into the interpreter is closed: __class__, __globals__, __code__, __subclasses__, __dict__ and the rest raise, though the operator and context-manager dunders you would implement on your own classes are fine. Single-underscore names (_helper, self._items) are ordinary and allowed. You also cannot exit the process: SystemExit, KeyboardInterrupt and the other non-Exception classes are absent, so raise an ordinary Exception to signal failure.

Everything else is Python as you know it. Classes, closures, comprehensions, generators, decorators, dataclasses, try/except, match, f-strings, augmented assignment and print all work normally, and the allowed modules cover the arithmetic, collections, text and serialization work this environment is for. Doctests you write in a docstring are run under exactly this policy too, so they are subject to the same rules as the code around them.

A violation is reported to you as an error naming the construct, and the program does not run. Read it as a boundary, not a bug to work around.

Configure the compile-time policy.

Args:

policy: RestrictedPython compile_restricted policy to compile: under. Defaults to RestrictedPythonPolicy, which is the policy the runtime guards in this module are matched to; a replacement has to keep emitting the same guard calls to stay safe.

_abc_impl = <_abc._abc_data object>

_allowed_modules_section() → PromptSection[source]

The import allowlist, rendered from the constant that enforces it.

The docstring above can name _ALLOWED_MODULES but not list it, and a model that has to discover the allowlist by getting an ImportError on os spends a turn per guess.

Return type:: PromptSection

_restricted_globals(env: Mapping[str, Any]) → dict[str, Any][source]

The namespace restricted code runs in: RestrictedPython’s safe builtins (extended by _EXTRA_SAFE_BUILTIN_NAMES and the guarded __import__/getattr/hasattr), the guarded accessors compiled code calls into, then env layered on top.

Return type:: dict[str, Any]

call_system(harness_prompt: PromptSection, agent_prompt: PromptSection) → Any[source]

Add the import allowlist, then the class docstring the base rule adds.

Appending before delegating puts _allowed_modules_section immediately ahead of the section PromptInjectingInterpretation contributes, so the two arrive together however the rest of the stack is composed – which is what lets the class docstring refer to the allowlist by name.

Return type:: Any

compile(source: str | AST, filename: str, mode: str = 'exec', flags: int = 0, dont_inherit: bool = False, optimize: int = -1) → CodeType[source]

compile_restricted under this provider’s policy.

Takes source text as readily as an AST, since run_doctests routes doctest’s own example compilation – which is textual, and in single mode – through this operation, so a docstring’s examples are held to the same policy as the code they document.

Return type:: CodeType

eval(bytecode: CodeType, env: dict[str, Any]) → Any[source]

Evaluate bytecode in a guarded namespace, for its value alone.

The same namespace exec builds, but with no binding copy-back: the operation’s contract discards binding effects, and the only eval-mode construct that could bind – a scope-escaping walrus – is rejected by the callers that compile expressions, so there is nothing to propagate.

Return type:: Any

exec(bytecode: CodeType, env: dict[str, Any]) → None[source]

Execute bytecode in a guarded namespace, copying its bindings back.

Execution happens in a namespace holding the runtime guards, so env itself is never handed to restricted code; what the code binds is copied back afterwards by comparing value identities, which catches both new names and rebindings of seeded ones. The sandbox’s own furniture (the guards, __builtins__, the injected _print) is excluded, being no binding effect of the code.

This is also where a docstring’s doctests are executed: run_doctests routes doctest’s own exec through this operation, so the examples run in the same guarded namespace as the code they document – otherwise >>> __import__("os").system(...) in a synthesized docstring would execute with nothing restricting it at all.

Return type:: None

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function RestrictedPythonExecutor.call_system>, Operation(parse, (source: str, filename: str) -> ast.Module): <function RestrictedPythonExecutor.parse>, Operation(compile, (source: str | ast.AST, filename: str, mode: str = 'exec', flags: int = 0, dont_inherit: bool = False, optimize: int = -1) -> code): <function RestrictedPythonExecutor.compile>, Operation(eval, (bytecode: code, env: dict[str, typing.Any]) -> Any): <function RestrictedPythonExecutor.eval>, Operation(exec, (bytecode: code, env: dict[str, typing.Any]) -> NoneType): <function RestrictedPythonExecutor.exec>}

parse(source: str, filename: str) → Module[source]

Parse source, registering it under filename so it stays readable.

The linecache entry is what keeps inspect.getsource working for objects defined by generated code, which has no file behind it. Parsing itself is unrestricted – the policy is applied in compile, over the tree this produces.

Return type:: Module

policy: type[RestrictingNodeTransformer] | None = None

class effectful.handlers.llm.harness.execution.restricted.RestrictedPythonPolicy(errors: list[str] | None = None, warnings: list[str] | None = None, used_names: dict[str, bool] | None = None)[source]

RestrictedPython’s policy, relaxed where it rejects ordinary modern Python.

RestrictingNodeTransformer predates a good deal of the language a model writes today, and its rejections are not all security-carrying. Four in particular make it unusable as-is for synthesized code:

annotated assignments (total: int = 0) are rejected outright, and with them every @dataclass and every class-level field – while this library asks models for annotated code and type-checks what it gets;
any name starting with ``_``, so a helper called _solve fails to compile;
``nonlocal``, so a closure cannot update the variable it closes over;
``counts[k] += 1``, so nothing can accumulate into a container in place.

This subclass allows those four (plus type aliases and the dunder methods in _SAFE_DUNDER_ATTRS, so a model can define __len__/__repr__ on its own classes) and changes nothing else. Everything the sandbox actually rests on is inherited untouched: exec/eval calls, star imports, async, except*, match and any other unreviewed syntax are still rejected, and attribute access, subscripting and iteration are still rewritten to the guarded accessors that RestrictedEvalProvider installs.

_check_pattern_expression(node: AST) → None[source]

Check the names and attributes of an expression sitting in a pattern position, without rewriting it.

Every such expression is a literal or a dotted name, and is only ever compared against the subject – never bound – so checking it under the same policy as an ordinary read is enough.

Return type:: None

check_name(node: Any, name: str | None, allow_magic_methods: bool = False) → None[source]

Check names if they are allowed.

If allow_magic_methods is True names in ALLOWED_FUNC_NAMES are additionally allowed although their names start with _.

Return type:: None

visit_AnnAssign(node: AnnAssign) → Any[source]

Allow annotated assignment (x: int = 1).

It carries no capability a plain assignment doesn’t: there is a single target and no unpacking, and an attribute or subscript target is rewritten to the write guard by the visitors for those nodes, exactly as in visit_Assign. The annotation is just another expression.

Return type:: Any

visit_Attribute(node: Attribute) → Any[source]

a.b -> _getattr_(a, 'b'), a.b = c -> _write_(a).b = c.

Identical to the base transform except that the name is checked against _is_allowed_attribute rather than “does it start with an underscore”.

Return type:: Any

visit_AugAssign(node: AugAssign) → Any[source]

Allow augmented assignment to an item or attribute (counts[k] += 1).

The base allows it only for a plain name (rewritten to _inplacevar_) and rejects a[i] += x / a.b += x outright, which rules out most code that accumulates into a container. Left alone, those compile to a subscript/attribute read, the in-place operation, and a store – and the target still goes through the ordinary visit_Subscript/visit_Attribute rewriting, so the store lands on _write_(a) and an attribute name is still checked. The read is the one thing not routed through _getitem_/_getattr_; that costs nothing here, where RestrictedEvalProvider reads are unrestricted, but a policy paired with a restricting _getitem_ should not use this class.

Return type:: Any

visit_Match(node: Match) → Any[source]

Allow match; the subject is an ordinary expression.

Return type:: Any

visit_MatchAs(node: MatchAs) → Any[source]

case x: / case [1] as pair: / case _: – a binding plus an optional sub-pattern.

Return type:: Any

visit_MatchClass(node: MatchClass) → Any[source]

case Point(x=0, y=y): – the class is checked in place, and each keyword names an attribute read off the subject, so it gets the attribute policy.

Return type:: Any

visit_MatchMapping(node: MatchMapping) → Any[source]

case {"k": v, **rest}: – keys are checked in place (they are pattern expressions), sub-patterns are visited, rest is a binding.

Return type:: Any

visit_MatchOr(node: MatchOr) → Any[source]

case 1 | 2: – alternatives are themselves patterns.

Return type:: Any

visit_MatchSequence(node: MatchSequence) → Any[source]

case [a, b]: – sub-patterns are visited; see the note on unguarded iteration above.

Return type:: Any

visit_MatchSingleton(node: MatchSingleton) → Any[source]

case None: / case True: – a bare constant, nothing to check.

Return type:: Any

visit_MatchStar(node: MatchStar) → Any[source]

case [first, *rest]: – rest is a binding.

Return type:: Any

visit_MatchValue(node: MatchValue) → Any[source]

case 3: / case Color.RED: – checked, deliberately not rewritten.

Return type:: Any

visit_Nonlocal(node: Nonlocal) → Any[source]

Allow nonlocal.

Like the global the base already allows, it only rebinds a name in an enclosing scope of the same generated code – and check_name still governs which names those can be.

Return type:: Any

visit_TypeAlias(node: AST) → Any[source]

Allow type X = ... aliases: a binding of a lazily-evaluated annotation expression, with no more reach than the expression itself.

Return type:: Any

visit_match_case(node: match_case) → Any[source]

Allow a case: its guard and body are ordinary code (and are rewritten as such); its pattern dispatches to the visitors below.

Return type:: Any

effectful.handlers.llm.harness.execution.restricted._ALLOWED_MODULES: Final = frozenset({'abc', 'array', 'base64', 'binascii', 'bisect', 'calendar', 'cmath', 'collections', 'collections.abc', 'copy', 'csv', 'dataclasses', 'datetime', 'decimal', 'difflib', 'enum', 'fractions', 'functools', 'graphlib', 'hashlib', 'heapq', 'itertools', 'json', 'math', 'numbers', 'operator', 'queue', 'random', 're', 'statistics', 'string', 'struct', 'textwrap', 'typing', 'unicodedata', 'uuid'})

Modules generated code may import.

Pure computation and data structures only: nothing that reaches the filesystem, the network, the process, or the import system itself (os, sys, subprocess, socket, importlib, builtins, inspect, types, pickle, ctypes, …). Submodules must be listed in full, since the check is on the imported name.

This list governs every module that reaches generated code, not just the ones it names in an import statement. An allowed module holds references to the modules it imported itself – uuid.os, typing.sys, queue.threading, json.codecs – so gating only __import__ would leave import uuid one attribute away from os.system. _guarded_getattr and _guarded_import therefore both refuse to hand back a module that is not named here, whichever module it was reached through.

Two near-misses worth recording, so they don’t get added later by analogy: io is not here because io.open is builtins.open, which would hand back the filesystem that omitting open closes; and numpy/matplotlib are not here because numpy.load(..., allow_pickle=True) executes arbitrary code. Synthesized code that genuinely needs those belongs under ~effectful.handlers.llm.harness.execution.builtin.BuiltinExecutor, not behind a widened allowlist.

effectful.handlers.llm.harness.execution.restricted._EXTRA_SAFE_BUILTIN_NAMES: Final = frozenset({'NotImplemented', 'RecursionError', 'StopAsyncIteration', 'TimeoutError', 'all', 'any', 'ascii', 'bin', 'bytearray', 'classmethod', 'dict', 'enumerate', 'filter', 'format', 'frozenset', 'iter', 'list', 'map', 'max', 'memoryview', 'min', 'next', 'object', 'property', 'reversed', 'set', 'staticmethod', 'sum', 'super', 'type'})

Builtins beyond RestrictedPython’s deliberately minimal safe_builtins.

Each is pure and reaches nothing outside its arguments; the omissions are the point – open, input, compile, eval, exec, globals, locals, vars, dir and breakpoint are all absent, and __import__/getattr/setattr/delattr are installed as guarded wrappers rather than taken from builtins.

effectful.handlers.llm.harness.execution.restricted._FORMAT_REACHING_CHARS: Final = ('.', '[')

Characters that make a str.format field reach past its argument.

They are matched against the field name, the part before any !conversion or :spec. A field that reaches – {0.__class__}, {0[1]} – is the whole reason RestrictedPython refuses these methods, since the traversal happens inside CPython’s formatter, where the _getattr_/_getitem_ rewriting cannot see it.

effectful.handlers.llm.harness.execution.restricted._GUARD_NAMES: Final = frozenset({'__builtins__', '__metaclass__', '_apply_', '_getattr_', '_getitem_', '_getiter_', '_inplacevar_', '_iter_unpack_sequence_', '_print', '_print_', '_unpack_sequence_', '_write_'}): Names the RestrictedPython transformer itself emits into compiled code – the guarded accessors (_getattr_(x, "y")), the print collector (_print), the class metaclass. Generated code must not bind or read them, or it could hand itself an unguarded accessor (or quietly disable one).

effectful.handlers.llm.harness.execution.restricted._INPLACE_OPS: Final[Mapping[str, Callable[[Any, Any], Any]]] = mappingproxy({'+=': <built-in function iadd>, '-=': <built-in function isub>, '*=': <built-in function imul>, '/=': <built-in function itruediv>, '//=': <built-in function ifloordiv>, '%=': <built-in function imod>, '**=': <built-in function ipow>, '<<=': <built-in function ilshift>, '>>=': <built-in function irshift>, '&=': <built-in function iand>, '^=': <built-in function ixor>, '|=': <built-in function ior>, '@=': <built-in function imatmul>})

The augmented-assignment operators, by symbol.

n += 1 compiles to n = _inplacevar_("+=", n, 1), so the environment has to supply the operators; without this every augmented assignment is a NameError.

effectful.handlers.llm.harness.execution.restricted._RAISE: Final = <object object>: _guarded_getattr’s sentinel default, distinguishing getattr(o, n) from getattr(o, n, None): absent a real default, a missing attribute raises.

effectful.handlers.llm.harness.execution.restricted._SAFE_DUNDER_ATTRS: Final = frozenset({'__abs__', '__add__', '__bool__', '__call__', '__contains__', '__doc__', '__enter__', '__eq__', '__exit__', '__ge__', '__getitem__', '__gt__', '__hash__', '__init__', '__iter__', '__le__', '__len__', '__lt__', '__mul__', '__name__', '__ne__', '__neg__', '__next__', '__post_init__', '__radd__', '__repr__', '__reversed__', '__rmul__', '__setitem__', '__str__', '__sub__'})

Dunder attributes generated code may read, call and define: the operator/context protocols plus a few purely descriptive names.

Every other __dunder__ – __class__, __bases__, __subclasses__, __mro__, __globals__, __code__, __dict__, __getattribute__, __reduce__, … – is the road from sandboxed code back to the interpreter, and stays closed.

effectful.handlers.llm.harness.execution.restricted._SAFE_DUNDER_BINDINGS: Final = frozenset({'__all__', '__slots__'})

Dunder bindings a class or module body may make.

Binding these declares something about the code being written; it is not a way to read a dunder attribute off an object, which _is_allowed_attribute governs separately.

class effectful.handlers.llm.harness.execution.restricted._StdoutPrintCollector(_getattr_=None)[source]

_print_ factory whose print(…) writes to the real sys.stdout (so output-capturing callers see it) rather than accumulating into the collector’s discarded printed buffer.

_call_print(*objects, **kwargs)[source]: Write straight to sys.stdout, defaulting file so an explicit print(..., file=x) in generated code still goes where it asked.

effectful.handlers.llm.harness.execution.restricted._UNSAFE_BUILTIN_NAMES: Final = frozenset({'BaseException', 'BaseExceptionGroup', 'GeneratorExit', 'KeyboardInterrupt', 'SystemExit'})

Builtins safe_builtins supplies that this environment takes back out: every BaseException subclass that is not an Exception.

raise SystemExit(...) in generated code unwinds straight through the harness and out of the host program – call_tool catches Exception, as it must, so that a real Ctrl-C still interrupts the process rather than being reported to the model as a failed tool call. That makes these four classes the one thing a sandboxed snippet can raise that its caller will not contain, and none of them is any use to the pure computation this environment is for: Exception and its subclasses remain available in full.

effectful.handlers.llm.harness.execution.restricted._checked_module(value: Any) → Any[source]

Pass value through, unless it is a module outside _ALLOWED_MODULES.

The allowlist is a statement about which modules generated code may hold, not merely which ones it may name in an import. Every allowed module carries the modules it imported itself as ordinary public attributes – and a public attribute is exactly what _guarded_getattr hands over without further question – so without this check uuid.os, typing.sys and queue.threading all read straight out of the sandbox.

The residual: this catches a module fetched as an attribute or imported, which is how modules are actually reached. A module returned from a call would not be checked – but nothing in _ALLOWED_MODULES returns one.

Return type:: Any

effectful.handlers.llm.harness.execution.restricted._checked_str_format(target: Any, name: str) → Callable[[...], str][source]

str.format/str.format_map, wrapped to check the template first.

RestrictedPython refuses these outright because "{0.__class__.__mro__}" walks attributes inside CPython’s formatter, out of reach of the guards – but refusing them wholesale is expensive and, worse, invisible: the call compiles and type-checks and only fails when the code runs, possibly inside a doctest. What is actually dangerous is the traversal, and that is written in the template, where it can be read off before anything is formatted. So check the template and allow the rest, which is nearly all real uses.

Return type:: Callable[..., str]

effectful.handlers.llm.harness.execution.restricted._guarded_apply(func: Callable, *args: Any, **kwargs: Any) → Any[source]

f(*args, **kwargs) – the form the transformer routes starred calls through. Argument values need no guarding here: whatever built them was itself compiled under the same policy.

Return type:: Any

effectful.handlers.llm.harness.execution.restricted._guarded_delattr(obj: Any, name: str) → None[source]

delattr under _is_allowed_attribute; see _guarded_setattr.

Return type:: None

effectful.handlers.llm.harness.execution.restricted._guarded_getattr(obj: Any, name: str, default: Any = <object object>) → Any[source]

The runtime half of _is_allowed_attribute: the _getattr_ every attribute access in restricted code compiles to, and the getattr builtin generated code sees (so a dynamically-computed name is checked too).

Return type:: Any

effectful.handlers.llm.harness.execution.restricted._guarded_hasattr(obj: Any, name: str) → bool[source]

hasattr that agrees with _guarded_getattr: a restricted attribute reads as absent rather than as a way to probe for one.

Return type:: bool

effectful.handlers.llm.harness.execution.restricted._guarded_import(name: str, globals: Any = None, locals: Any = None, fromlist: Sequence[str] = (), level: int = 0) → ModuleType[source]

The __import__ restricted code sees: _ALLOWED_MODULES only.

Without an __import__ at all, import math fails and most generated code with it; with the real one, import os hands back the process. Relative imports are refused outright – there is no package for generated code to be relative to.

Return type:: ModuleType

effectful.handlers.llm.harness.execution.restricted._guarded_inplacevar(op: str, x: Any, y: Any) → Any[source]

x <op>= y, as _INPLACE_OPS spells it.

Return type:: Any

effectful.handlers.llm.harness.execution.restricted._guarded_setattr(obj: Any, name: str, value: Any) → None[source]

setattr under _is_allowed_attribute, matching what obj.name = value compiles to (RestrictedPython checks that name at compile time; this is the same rule for a name computed at run time).

Return type:: None

effectful.handlers.llm.harness.execution.restricted._is_allowed_attribute(name: str) → bool[source]

Whether generated code may read, write or define the attribute name.

Same trade as _is_allowed_name, one level in: single-underscore attributes (self._items) are ordinary Python, while dunders outside _SAFE_DUNDER_ATTRS – and the frame/code/traceback attributes RestrictedPython lists in INSPECT_ATTRIBUTES – are how sandboxed code climbs out.

Return type:: bool

effectful.handlers.llm.harness.execution.restricted._is_allowed_name(name: str) → bool[source]

Whether generated code may bind or read name as an identifier.

RestrictedPython rejects every name starting with _, which costs a great deal (_helper, _memo, _solve are how Python spells “private”) and buys little: an identifier is only dangerous when it is one of the guard names the transformer emits, or a dunder. Both of those stay rejected; a plain single-underscore private does not.

Return type:: bool

effectful.handlers.llm.harness.execution.restricted._reject_reaching_format_fields(template: str) → None[source]

Raise unless every replacement field in template names its argument and stops there – no attribute or item traversal from it.

Nested fields inside a format spec ("{0:{1}}") are checked too; the recursion terminates because a spec with no braces yields a single field of None, and CPython caps spec nesting at one level regardless.

Return type:: None

Validation

Operations and handlers that type-check and doctest model-authored Python before it runs.

effectful.handlers.llm.harness.validation.hooks.run_doctests(obj: Callable | type | ModuleType, globs: Mapping[str, Any]) → None[source]

Run the doctests found in a synthesized object’s docstring.

Return type:: None

obj: The synthesized object (typically a function) whose docstring may: contain interactive >>> examples.
globs: The namespace the examples execute in (typically the exec namespace,: which already contains the function plus its lexical context).

Returns None, raises TypeError if any doctest example fails. A docstring with no examples is a no-op (passes trivially).

Unlike the other operations here, this one carries its mechanics in its default rule: finding the examples and reporting their failures is the same work whatever provider is installed. What differs is how each example is compiled and executed, and that is delegated to the compile and exec operations – so a docstring’s examples run under exactly the provider that runs the code they document, and no provider at all is an error here just as it is there.

effectful.handlers.llm.harness.validation.hooks.type_check(source: str, lo: int | None = None, hi: int | None = None, *, lenient: bool = False) → None[source]

Type check a module source, reporting only diagnostics inside a line region.

Return type:: None

source: A complete module source to check (e.g. produced by: splice_into_source, which splices generated code into a Skill’s real module source).
lo, hi: Inclusive line range within source to report errors from; when: omitted, the whole source is in scope. Errors outside the region are ignored so unrelated pre-existing code never blocks synthesis.
lenient: when True, relax the check for incrementally-built REPL code spliced into: a Skill body – allow redefinition (a cell may rebind or redefine a name) and don’t require the body to satisfy the Skill’s return type. Off (strict) for a synthesized Callable or SkillBody, which must honor its signature and gets no redefinition slack. How much slack this buys is up to the handler: it is a list of disabled mypy error codes under MypyTypeChecker and close to a no-op under TyTypeChecker, which is already this permissive.

Returns if the source type-checks, raises TypeError on an in-region failure.

Unlike parse/compile/exec, which have no meaning without a provider, type checking is an optional layer over them: the default rule below passes everything, so a stack with no type checker installed runs generated code unchecked rather than refusing to run it at all. MypyTypeChecker and TyTypeChecker are the handlers that make the check real.

Enforcement of the pre-conditions a caller writes into a Skill’s parameters.

A parameter annotated with pydantic metadata – a pydantic.AfterValidator, an annotated_types constraint, a pydantic.Field – states a contract its argument must satisfy:

@Skill.define
def select_seat(user_input: Annotated[str, Predicate(is_seat_request)]) -> Seat:
    """Extract the seat from {user_input}."""

What PydanticSkillArgValidator changes is the top-level call – a skill invoked from Python, by a caller who wrote the annotation and can reasonably expect it to mean something. Nothing in Python consults an annotation, so without this handler the contract silently lapses there. Installed, the argument is validated before the prompt is built, and a violation raises pydantic.ValidationError – a ValueError, so ordinary handling still applies – rather than reaching the model at all.

A skill the model calls as a tool is validated with or without this handler: call_assistant decodes each argument through Encodable of the parameter’s own annotation, so the metadata is applied as the call is decoded, before the skill is ever entered. Two consequences follow.

For a model-supplied argument the validation happens twice – once at decode, once here – so a pre-condition that costs something (an LLM-backed predicate, say) pays it twice.
The exception is the expression pathway (~effectful.handlers.llm.harness.synthesis.toolcall.ExpressionToolCaller), which evaluates a Python call expression instead of decoding JSON arguments and does not apply the parameter’s metadata. There this handler is the only thing enforcing the contract, which is why it is installed for every tool-calling mode rather than only the JSON one.

Post-conditions need no handler

The mirror image – metadata on the return annotation – is enforced by the decoder itself, since call_assistant decodes an answer through Encodable of the skill’s declared return type and the caller’s metadata rides along. A rejection there is a decoding failure, which ~effectful.handlers.llm.harness.durability.retrying.TenacityRetryer feeds back to the model as the instruction for its next attempt. So this handler governs the way in; the way out is contracted whether or not it is installed.

class effectful.handlers.llm.harness.validation.pydantic.PydanticSkillArgValidator[source]

The arguments you were given have already been checked. Where a parameter’s annotation carries a constraint – a value range, a pattern, a predicate that has to hold – that constraint was evaluated before this call reached you, and an argument that failed it never got here.

So do not re-verify them. Spending a turn confirming that an argument satisfies a condition it was admitted for is work the harness already did, and its result cannot differ. Take the arguments as given and answer the request.

Your answer is checked the same way, against any constraint on the return annotation. If it fails, you get the validation error back and another attempt, so an answer that is close but out of the declared range is worth correcting before you send it.

call_agent(skill: Skill[P, T], *args: P, **kwargs: P) → T[source]

Validate the annotated arguments, then forward the normalized call.

Only a parameter carrying metadata of its own is touched, so a skill that declares no contracts is unaffected: nothing is validated, and no argument is round-tripped through the encoding (which would copy it). Variadic parameters are validated element-wise, since the metadata describes each item rather than the tuple or dict collecting them.

The validated value replaces the bound one, so a validator that normalizes is applied rather than consulted and discarded, and it is the normalized arguments that get forwarded and rendered into the prompt.

Validation runs under the call environment, so a pre-condition may be stated relative to the rest of the call – info.context holds the other arguments and the skill’s lexical scope, exactly as it does when the answer is decoded on the way back.

Return type:: TypeVar(T)

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function PydanticSkillArgValidator.call_agent>}

Type checking of generated code by shelling out to mypy.

MypyTypeChecker is independent of any executor: it says how generated code is checked, not how it is parsed, compiled or run, so it is installed alongside whichever of those handlers a stack uses:

handler(MypyTypeChecker()), handler(BuiltinExecutor())

rather than being part of one.

It is interchangeable with ~effectful.handlers.llm.harness.validation.ty.TyTypeChecker, which implements the same operation with the same contract and is substantially faster; that module’s docstring compares the two.

class effectful.handlers.llm.harness.validation.mypy.MypyTypeChecker(lenient_flags: tuple[str, ...] = ('--allow-redefinition-new', '--local-partial-types', '--disable-error-code=no-redef', '--disable-error-code=return', '--disable-error-code=empty-body')) → None[source]

Python you write is type-checked before it is run, by mypy. Code that fails the check does not execute at all: you get mypy’s diagnostics back – the message, the error code, the offending line – and the turn is yours again to fix them.

Treat that as a fast, free reviewer rather than an obstacle. Annotate what you write, use the types the surrounding code declares, and read a diagnostic as a claim about your code that is usually correct. Silencing one with typing.Any or a blanket # type: ignore will pass the check and then fail at runtime, where the error costs a whole turn instead of none.

Only the code you generate is checked; errors elsewhere in the module you are working in are not yours to fix and will not block you.

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(type_check, (source: str, lo: int | None = None, hi: int | None = None, *, lenient: bool = False) -> NoneType): <function MypyTypeChecker.type_check>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

lenient_flags: tuple[str, ...] = ('--allow-redefinition-new', '--local-partial-types', '--disable-error-code=no-redef', '--disable-error-code=return', '--disable-error-code=empty-body'): Flags added under lenient=True to waive the diagnostics that a REPL transcript or a spliced function body provokes by construction – see type_check for what each one is for.

type_check(source: str, lo: int | None = None, hi: int | None = None, *, lenient: bool = False) → None[source]

Run mypy on source and raise TypeError if any error diagnostic falls within [lo, hi]; raise RuntimeError if mypy itself fails to run.

Applies mypy to whatever source it’s given – spliced or otherwise – and reports only the region’s errors (the whole source when the region is omitted), so pre-existing errors elsewhere in source never block synthesis.

When lenient (for REPL code spliced into a Skill body): allow a variable to be redefined with a new type across cells (--allow-redefinition-new, which supersedes the narrower --allow-redefinition and requires --local-partial-types), a def/class/import to be redefined (no-redef), and the body not to return the Skill’s declared type (return/empty-body). All normal for an incrementally-built REPL, not real errors.

Return type:: None

Type checking of generated code by shelling out to ty.

TyTypeChecker is interchangeable with ~effectful.handlers.llm.harness.validation.mypy.MypyTypeChecker – same operation, same contract, a different checker behind it. ty is a compiled binary that needs no per-call cache and builds no module graph in this process, so a check costs milliseconds where mypy’s costs seconds, and on the failure path it reports the offending line with ty’s own hints rather than a line of JSON. Prefer it unless a stack specifically needs mypy’s analysis.

Either checker is independent of any executor: it says how generated code is checked, not how it is parsed, compiled or run, so it is installed alongside whichever of those handlers a stack uses:

handler(TyTypeChecker()), handler(BuiltinExecutor())

rather than being part of one.

class effectful.handlers.llm.harness.validation.ty.TyTypeChecker(lenient_ignored_rules: tuple[str, ...] = ('conflicting-declarations',)) → None[source]

Python you write is type-checked before it is run, by the ty type checker. Code that fails the check does not execute at all: you get ty’s diagnostics back – the message, the offending line, its hints – and the turn is yours again to fix them.

Treat that as a fast, free reviewer rather than an obstacle. Annotate what you write, use the types the surrounding code declares, and read a diagnostic as a claim about your code that is usually correct. Silencing one with typing.Any or a blanket # type: ignore will pass the check and then fail at runtime, where the error costs a whole turn instead of none.

Only the code you generate is checked; errors elsewhere in the module you are working in are not yours to fix and will not block you.

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(type_check, (source: str, lo: int | None = None, hi: int | None = None, *, lenient: bool = False) -> NoneType): <function TyTypeChecker.type_check>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

lenient_ignored_rules: tuple[str, ...] = ('conflicting-declarations',): Rules ignored under lenient=True. Deliberately short: ty already grants most of that leniency unasked – see type_check.

type_check(source: str, lo: int | None = None, hi: int | None = None, *, lenient: bool = False) → None[source]

Run ty on source and raise TypeError if any error diagnostic falls within [lo, hi]; raise RuntimeError if ty itself fails to run.

Applies ty to whatever source it’s given – spliced or otherwise – and reports only the region’s errors (the whole source when the region is omitted), so pre-existing errors elsewhere in source never block synthesis.

lenient disables far less here than under MypyTypeChecker, because ty grants most of that leniency unasked. A variable may be redefined with a new type across cells and ty narrows to the latest binding, and a def/class/import may be redefined – no flag needed for either. A body that doesn’t return the Skill’s declared type is reported against the signature line, while a body that returns the wrong type is reported against the return statement, so the region filter tells those two apart on its own; splitting them by position rather than by flag is what keeps lenient from also waiving a genuine wrong-return-type error. That leaves no-redef’s counterpart, kept for faithfulness to ty’s own mapping though it fires on none of the redefinition shapes a REPL produces.

Return type:: None

Synthesis

Handlers that let the model answer with code rather than data.

A stateful REPL, a synthesized function, a synthesized body for the calling Skill, and expression-based tool calls (the pathway that supports polymorphic tools).

A persistent, stateful Python REPL offered to the model as a tool.

StatefulReplSynthesizer is off by default; install it where the LLM should be able to run code whose state – variables, imports, definitions – survives across tool calls within a single Skill invocation.

Scoping mirrors how __history__ is managed for Skill calls: call_agent introduces fresh session-bound handlers (exec_code, repl_history, repl_env) for the duration of the call, and call_assistant injects the tool routed to that session. The session is therefore introduced and eliminated by its own handler, bounded to the Skill call by construction – there is no global registry of sessions, and nested Skill calls get their own isolated ones.

That bound is the fact the model most needs and is least able to infer. It sees one conversation, in which an Agent’s earlier calls are still visible as earlier user messages, and nothing in the transcript distinguishes “a session opened here” from “a turn happened here”. So the handler states it twice: once in general, in the class docstring that becomes its system-prompt section, and once concretely, in the REPL session section call_user attaches to every request – which is also where the call’s arguments are claimed as session bindings, since they appear in no table of the system message.

The session is seeded from the Skill’s lexical context and routes execution through the parse/compile/exec effect operations, so it works under any installed eval provider (~effectful.handlers.llm.harness.execution.builtin.BuiltinExecutor or ~effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor).

class effectful.handlers.llm.harness.synthesis.snippet.ReplSession(env: MutableMapping[str, Any])[source]

A persistent, output-capturing Python session seeded from a lexical context.

exec_code(source) runs a pre-compiled code object in self.locals through the exec effect operation. Both bindings and captured stdout/stderr persist across calls – variables, imports and definitions accumulate exactly like a REPL – and the session (with its buffer) is discarded as a whole when it goes out of scope. Each call returns only the output it produced; there is no bare-expression auto-echo, so use print() to surface values.

Constructor.

The optional ‘locals’ argument specifies a mapping to use as the namespace in which code will be executed; it defaults to a newly created dictionary with key “__name__” set to “__console__” and key “__doc__” set to None.

exec_code(code: CodeType) → str[source]

Run code in this session’s namespace and return what it printed.

Every snippet runs in the SAME namespace, so imports, definitions and assignments accumulate: this is a REPL, not a one-shot sandbox. The namespace starts seeded from the enclosing Skill call’s scope (its bound arguments over the Skill’s lexical context), which a snippet may read and rebind. Its lifetime is that call’s: see StatefulReplSynthesizer.call_agent, which creates and discards it, and the class docstring there for why the model is told so twice.

Returns this snippet’s own slice of the session’s output – stdout (what print wrote) then stderr. There is no bare-expression auto-echo, so a snippet that prints nothing returns the empty string.

A snippet that raises propagates; the session and every binding made before the raise survive, so the next snippet can repair it. Output printed before the raise is not returned, since the raise reaches the caller in its place.

Return type:: str

locals: dict[str, Any]

property prior_snippets: list[str]: Sources of the actual error-free executed snippets, in order – the type-check context the Encodable[CodeType] decoder splices before the current snippet.

runcode(code: CodeType) → None[source]

Execute a code object.

When an exception occurs, self.showtraceback() is called to display a traceback. All exceptions are caught except SystemExit, which is reraised.

A note about KeyboardInterrupt: this exception may occur elsewhere in this code, and may not always be caught. The caller should be prepared to deal with it.

Return type:: None

stderr: StringIO

stdout: StringIO

class effectful.handlers.llm.harness.synthesis.snippet.StatefulReplSynthesizer[source]

You may run arbitrary Python code in a persistent session, through the exec_code tool. Its own description states how it behaves — the namespace it runs in, what it returns, what it will reject — and that description is authoritative; follow it rather than any recollection of how such a tool usually works. To see the value of anything in scope, print it.

## Writing onto self

The REPL session itself lasts one call, so anything you want to keep has to go somewhere the program owns rather than somewhere the session owns — self is the usual place, and any other in-scope object works the same way. A value, a note, or a function you define here:

```python # this call def next_guess(prefix: str) -> str:

return prefix + “A”

self.next_guess = next_guess # a bound function is offered as a tool later self.codes[“room0”] = “BBA” # and a plain value is just there ```

`python # a later call, new session, `next_guess` and `codes` still on self print(self.codes["room0"]) `

call_agent(skill: Skill[P, T], *args: P, **kwargs: P) → T[source]

Open a REPL session for the duration of this call.

The session’s namespace is seeded from a collections.ChainMap of the bound arguments over the Skill’s lexical context, so this call’s own arguments shadow same-named module globals – which is the scope the model is shown, and the scope a snippet executes against.

Every session-scoped operation is bound here, inside the handler block, so the session is created and discarded with the call. A nested Skill call runs this rule again and gets a namespace of its own; nothing outlives the with.

Return type:: TypeVar(T)

call_assistant(messages: Sequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], response_type: type[T], env: Mapping[str, Any], tools: Set[Tool] = frozenset({})) → AssistantResult[source]

Offer the REPL tools, over an env that includes the session namespace.

The session’s bindings are merged over the request env, so a name the model defined in an earlier snippet is visible to whatever the next request decodes – notably the type check a new snippet is held to, which would otherwise see every such name as undefined.

Return type:: GenericAlias[TypeVar(T)]

call_tool(tool_call: DecodedToolCall) → ToolResult[source]

Compact the conversation after a successful exec_code(compact=...).

Only on success: a snippet that raised clears nothing, since the model has not yet had the chance to record what mattered, and the traceback it needs to repair the snippet is in the very round a clear would drop.

Forwarding first is what makes this safe to do inline. By the time control returns, HistoryBuilder has appended this call’s tool message, so the round compact keeps is complete; and because compaction only removes messages ahead of the assistant message that requested this call, a sibling tool call answered later still finds that message and cannot be orphaned.

Return type:: GenericAlias[TypeVar(T)]

call_user(user_prompt: PromptSection) → Any[source]

Tell each request which session it opens, and what is bound in it.

Attached to the request rather than to the system message because both facts are per-call: the session boundary is this message, and the arguments named by its heading are this call’s.

The section is the same text every time, so this rule needs nothing from the call it decorates – which is why it can be a rule at all, rather than something call_agent closes over the Skill to build.

Return type:: Any

final classmethod exec_code(compact: CompactionScope = CompactionScope.NONE) → str[source]

Run Python in a stateful session and return its output.

This is a REPL, not a one-shot sandbox: every call within THIS Skill call runs in the SAME namespace, so imports, definitions and assignments accumulate across your turns. The namespace is seeded from the surrounding scope – this call’s arguments and the Lexical scope table – which you may read and rebind. The session ends when you answer; the next request gets an empty one. Only self and the surrounding scope outlive it.

Your code is run, and type-checked, as if it were the body of the function you are answering for, so read the names already in scope instead of re-importing modules or retyping values the request gave you.

Output: returns this call’s output – its stdout (what print wrote) followed by anything written to stderr. There is NO automatic echoing of results – a bare expression on its own line (e.g. 1 + 1) displays nothing, so call print(…) for anything you want to see.

A snippet that raises comes back as a failed call carrying the traceback. The session and every binding made before the raise survive, so read the error, fix the code, and continue in the next call – but output printed before the error is not returned, so print again once it works.

compact compacts the conversation – see its own schema below for what each scope drops – but only if the snippet raises no exception: a snippet that raises compacts NOTHING.

Whichever scope you pick, the current request and THIS call of yours survive: your message, the snippet you wrote and its output. So the snippet and the message you send it with are your note to your later self, and you do not have to route everything through print(…) to keep it. What a compaction cannot save is what you never wrote down at all – and only self survives the next one, so anything that must last belongs there.

Return type:: str

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_tool, (tool_call: effectful.handlers.llm.harness.serialization.DecodedToolCall[T]) -> ToolResult[T]): <function StatefulReplSynthesizer.call_tool>, Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function StatefulReplSynthesizer.call_assistant>, Operation(call_user, (user_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionUserMessage): <function StatefulReplSynthesizer.call_user>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function StatefulReplSynthesizer.call_agent>}

final classmethod repl_env() → dict[str, Any][source]

The REPL session’s current namespace, as a flat dict of name -> value

Return type:: dict[str, Any]

final classmethod repl_history() → list[str][source]

This REPL session’s error-free executed snippets, in order.

Empty by default: unlike the tool operations above, this one is asked for by a decoder (Encodable[CodeType], to type-check a snippet against the session it will run in), which can be reached with no REPL in scope at all – decoding a code object outside a managed StatefulReplSynthesizer call. “No session” is a meaningful answer there (no prior snippets), not a missing handler.

Return type:: list[str]

type effectful.handlers.llm.harness.synthesis.function.SplicedRegion = tuple[str, int, int]

class effectful.handlers.llm.harness.synthesis.function.SynthesizedFunction(**data: Any) → None[source]

Structured output for function synthesis.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

code: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Answering a Skill by synthesizing its body.

This is the declarative “CodeAdapt” workflow: the LLM writes code implementing the body of the Skill rather than reasoning out the answer itself. FinalBodySynthesizer offers the synthesis tool alongside the Skill’s normal completion paths rather than replacing them – across turns the model may freely call any other tool in scope (their results are fed back as usual), and it may still answer the return type directly via structured output. The loop terminates when it either answers directly or calls write_and_run_body. To force the synthesis path, pass tool_choice="required"; handler config is forwarded to the model request.

The function is synthesized by reusing the existing Callable synthesis machinery: the tool’s argument is typed as Callable[[params], ret], so call_assistant’s tool-call decoding parses, type-checks, compiles and executes the model’s code into a real function before it is applied. An eval provider (~effectful.handlers.llm.harness.execution.builtin.BuiltinExecutor or ~effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor) must therefore be installed.

Failures compose with ~effectful.handlers.llm.harness.durability.retrying.TenacityRetryer: a function that fails to synthesize surfaces as a ToolCallDecodingError, and one that raises when applied to the inputs as a ToolCallExecutionError; both are fed back to the model as a tool message and the loop continues so it can revise:

with (
    handler(AgentLoop()),
    handler(LiteLLMConfigurer(model="gpt-5-mini")),
    handler(HistoryBuilder()),
    handler(FinalBodySynthesizer()),
    handler(TenacityRetryer()),
):
    ...

class effectful.handlers.llm.harness.synthesis.body.FinalBodySynthesizer[source]

You can state a Skill’s answer directly, or you can compute it by writing an implementation and submitting it with the write_and_run_body tool. This section is about the tool. Reach for it when working the answer out by hand would be error-prone — a search, an enumeration, a constraint to check against — or when the Skill’s doctests are the standard your answer has to meet.

A direct answer is also accepted, and is the right choice when you already hold the value: do not wrap a value you have in hand inside a function that ignores its arguments and returns a constant.

The tool’s own description says what to submit and what the code must satisfy; follow it rather than any recollection of how such a tool usually works. Two things it does not tell you. Your function may reference names from the lexical scope (see the Lexical scope table). And what your submission is judged on is the Skill’s doctests: the harness attaches the Skill’s docstring to your function and runs its examples, with recursive calls to the Skill routed back to your implementation. A solution whose doctests fail — or that raises when applied — is rejected and returned to you to revise, so the answer only stands once those examples pass.

A Skill whose declared return type is itself a function is a different thing, easily confused with this one: you answer it by writing the function it returns, as an ordinary direct answer, and this tool is not involved. Three rules invert there, and nothing else states them. The signature to write is the returned function’s, taken from the return type — not the Skill’s own, and with no self receiver even when the Skill is a method. Every parameter and the return type must be annotated there, where for this tool they are optional. And your docstring is kept rather than replaced, so if the Skill asks for doctests certifying what you wrote, write them: they are run, and they are what your answer is accepted on.

A successful write_and_run_body call ends the call immediately: no further turn is taken, and the value of applying your function to the original arguments is the Skill’s answer. Because it ends the call, it must be the only tool call in its turn — call any other tools you need on earlier turns, and call write_and_run_body by itself once you are ready to answer.

This answers the current call only. A submission is not a standing answer: if an earlier user message in this conversation was a previous call that you answered this way, that answer has already been returned to the program and has nothing to do with the question you are being asked now. To answer this call by synthesis you must call write_and_run_body again.

call_agent(skill: Skill[P, T], *args: P, **kwargs: P) → T[source]

Offer write_and_run_body for the duration of this call.

The tool is built per call, closing over these bound arguments: it is what applies the submitted function to them, so it cannot be shared across calls. It is added by an inner call_assistant rule rather than up front, because the request’s tool set is assembled further down the stack and only exists once the assistant is actually being called.

The guard makes the injection idempotent: a nested call reaching this rule again with the tool already in the set forwards unchanged, so a recursive Skill is not offered several generations of its own submission tool.

Return type:: TypeVar(T)

call_tool(tool_call: DecodedToolCall) → Any[source]

Mark a successful write_and_run_body call as the Skill’s answer, and honour the compact it was submitted with.

This is the rule that terminates the completion loop on the synthesis path: the model is free to answer directly instead, and every other tool call forwards untouched, so setting is_final here is the only thing that distinguishes a submission from an ordinary tool result.

Only a successful one: when TenacityRetryer captures a submission that raised – the synthesized function errored on the real arguments – it hands back the ToolCallExecutionError in place of a result, and that is no answer to finalize on. Leaving is_final alone there is what gives the model the next turn to revise, and compacting there would throw away the error it needs to do so.

The compaction runs here rather than anywhere later because there is no later: this call ends the loop. compact keeps the round that asked for it, which is what makes that safe – the surviving history ends on this submission and its result rather than on a request nobody answered, and the source the model submitted stays in the assistant message’s arguments where it can read it back on the next call.

Return type:: Any

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_tool, (tool_call: effectful.handlers.llm.harness.serialization.DecodedToolCall[T]) -> ToolResult[T]): <function FinalBodySynthesizer.call_tool>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function FinalBodySynthesizer.call_agent>}

class effectful.handlers.llm.harness.synthesis.body.MethodSkillBody[source]

A SkillBody for an instance-method Skill.

Carries the method/free distinction on the type’s origin (context-free schema generation reads it) so write_and_run_body’s description names the leading receiver self and the receiver is exempt from the annotation requirement – the model no longer has to reverse-engineer that the first parameter is self. The Skill’s real signature (which includes the receiver) remains the type-check contract; see splice_skill_body.

class effectful.handlers.llm.harness.synthesis.body.SkillBody[source]

The synthesized body of a Skill, as opposed to a general Callable.

Used only as the type of write_and_run_body’s implementation parameter (see effectful.handlers.llm.harness.synthesis.body.FinalBodySynthesizer). A SkillBody[[P], R] carries the Skill’s parameter and return types exactly like a Callable, but gets its own TypeToPydanticType case (_pydantic_skill_body) so the synthesized function is type-checked against the enclosing Skill’s source and its doctests run with self/recursive calls routed to the synthesized implementation. The enclosing Skill is recovered from the decode context (the anchor), so no state rides on the type itself.

class effectful.handlers.llm.harness.synthesis.body.SynthesizedMethodSkillBody(**data: Any) → None[source]

Structured output for synthesizing an instance-method Skill’s body.

Decoded through _pydantic_skill_body: the function is type-checked against the enclosing Skill’s source and its doctests are run with self/recursive calls routed to the synthesized implementation.

Unlike SynthesizedFunction, the parameter and return annotations are not required: a Skill body is type-checked against the Skill’s own signature (see splice_skill_body), so the model may omit or vary them – in particular it need not annotate the self receiver of an instance-method Skill.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

code: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class effectful.handlers.llm.harness.synthesis.body.SynthesizedSkillBody(**data: Any) → None[source]

Structured output for synthesizing a Skill’s body (write_and_run_body).

Decoded through _pydantic_skill_body: the function is type-checked against the enclosing Skill’s source and its doctests are run with self/recursive calls routed to the synthesized implementation.

Unlike SynthesizedFunction, the parameter and return annotations are not required: a Skill body is type-checked against the Skill’s own signature (see splice_skill_body), so the model may omit or vary them – in particular it need not annotate the self receiver of an instance-method Skill.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

code: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Calling lexically scoped tools by writing Python expressions.

ExpressionToolCaller is the code-generation transformation stage over the lexical tools an extractor (~effectful.handlers.llm.harness.legibility.lexical.LexicalToolExtractor or ImplicitToolExtractor) discovered (issues #489/#505): instead of leaving each lexical tool advertised with a JSON parameters schema – degenerate for a polymorphic tool, whose parameter types carry TypeVars – every lexical tool arriving in the request’s tools is replaced by a wrapper so the model must write a call expression. Decoding the expression does all the work up to the call itself (type check, callee check, argument evaluation; see _pydantic_type_call_expression), so a bad expression is a ToolCallDecodingError the retry loop feeds back. The decoded CallExpression IS a DecodedToolCall for the underlying tool; call_assistant substitutes it for the wrapper’s call, so call_tool and every handler of it see the real tool bound to real argument values.

MixedToolCaller is the default lexical tool caller, and narrows this to the tools that need it: schema-constrained JSON arguments where a schema can describe the tool faithfully, the expression pathway where it cannot. Everything but the partition predicate (MixedToolCaller._should_wrap) is inherited. A model that prefers writing code can still call any JSON tool from the REPL (exec_code) where ~effectful.handlers.llm.harness.synthesis.snippet.StatefulReplSynthesizer is installed.

Install either caller above an extractor – the extractor is the sole source of lexical tools, so a caller without one beneath it has nothing to wrap – and below anything else that contributes tools: tools arriving from other handlers pass through untouched, and the anchor Skill must not be wrapped (the default call_assistant rule subtracts it by identity, which cannot see through a wrapper), so it is excluded rather than wrapped. An eval provider (~effectful.handlers.llm.harness.execution.builtin.BuiltinExecutor or ~effectful.handlers.llm.harness.execution.restricted.RestrictedPythonExecutor) must be installed for the parse/compile/eval operations.

class effectful.handlers.llm.harness.synthesis.toolcall.CallExpression(tool: Tool[..., T], bound_args: BoundArguments, id: ToolCallID, name: str, source: str) → None[source]

A decoded tool-call expression: a DecodedToolCall for the underlying tool, remembering the source it was decoded from.

The decoded form of the call argument of an expression tool call (see ExpressionToolCaller). Decoding does everything except the call itself: the source is parsed, type-checked in the anchor Skill’s scope, its callee resolved (and checked against the tool the expression was submitted for), and its argument expressions evaluated – so what remains is exactly a decoded call to the underlying raw tool, and that is what this is: ExpressionToolCaller substitutes it (stamped with the outer call’s id/name) for the wrapper’s DecodedToolCall, so every call_tool handler – the default rule, retryers, tracers – sees the real tool bound to real argument values, and every way the model’s expression can be wrong surfaces as a decode error rather than an execution error.

source (optional on the base, required here) is the round-trip form: the advertised schema for an expression call is {"call": <source>}, and the evaluated bound_args may hold values with no JSON encoding at all.

class effectful.handlers.llm.harness.synthesis.toolcall.ExpressionToolCaller[source]

The tools of this Skill’s lexical scope are called by writing Python, not by filling in JSON arguments. Each one takes a single call parameter, and its own description gives the exact reference to invoke it by and the signature to match – read that description rather than guessing a name off the Lexical scope table.

Two things those descriptions do not tell you. The arguments inside a call may be arbitrary Python expressions over the names in the Lexical scope table and any bindings you have made in the REPL session, so you can write extend_sequence(examples, make_example()) rather than only literals. And the expression must be a single call to the advertised tool – no statements, no assignments, no := bindings (bind names with exec_code first if you need them).

call_assistant(messages: Sequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], response_type: type[T], env: Mapping[str, Any], tools: Set[Tool] = frozenset({})) → AssistantResult[source]

Wrap the lexical tools among tools, then unwrap the calls the model made to them.

The lexical tools arrive pre-populated by the extractor installed beneath this handler; which of tools count as lexical is decided by re-walking the scope (_tool_paths). Explicitly declared tools are found in the walk directly; an implicitly wrapped one is found through its __implicit_target__ – the env binds the raw callable, not the wrapper, so the walk’s wrap hook resolves each raw binding back to the wrapper in tools that stands for it (==, because a bound method is a fresh object on every attribute access). Each lexical tool _should_wrap selects is replaced by its expression wrapper (never offered alongside it); the rest pass through untouched: tools other handlers contributed (REPL access, synthesis, readers), lexical tools the JSON pathway can advertise faithfully (under MixedToolCaller), and the anchor Skill, which must stay recognizable by identity to the default rule that subtracts it.

On the way back, each wrapper call is replaced by the CallExpression it decoded – itself a DecodedToolCall, for the underlying tool bound to real argument values – stamped with the outer call’s id and name so the transcript still lines up with what the model sent. Every call_tool handler (the default rule, retryers, tracers) therefore sees the actual call rather than the wrapper’s. This handler is innermost, so the substitution happens before any of them run.

Return type:: GenericAlias[TypeVar(T)]

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function ExpressionToolCaller.call_assistant>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

class effectful.handlers.llm.harness.synthesis.toolcall.MixedToolCaller[source]

Most tools in this Skill’s lexical scope are ordinary tools: call them by name with JSON arguments matching their schema. Tools whose signatures a JSON schema cannot capture – generic (type-variable) parameters, variadic *args/**kwargs, or parameter types with no JSON encoding – are instead called by writing Python, and take a single call parameter holding one Python expression that invokes them.

Which mode a tool uses is fixed by its signature, not by your preference, and its own description tells you which it is and what to write – read that rather than guessing from the Lexical scope table.

One thing those descriptions do not tell you: the arguments inside a call may be arbitrary Python expressions over the names in the Lexical scope table and any bindings you have made in the REPL session, so you can write extend_sequence(examples, make_example()) rather than only literals. The expression must still be a single call to the advertised tool – no statements, no assignments, no := bindings (bind names with exec_code first if you need them).

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function ExpressionToolCaller.call_assistant>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

Durability

Handlers that make a call survive failure.

Message-history accumulation, transactional rollback, retrying on malformed model output, and checkpointing a persisted Agent to SQLite.

class effectful.handlers.llm.harness.durability.transaction.CompactionScope(*values)[source]

How much of the conversation a compacting tool call drops.

"none" compacts nothing. "turn" drops the current call’s earlier rounds, keeping every previous call. "conversation" additionally drops those previous calls, leaving the system message, the request and the asking round.

CONVERSATION = 'conversation'

NONE = 'none'

TURN = 'turn'

class effectful.handlers.llm.harness.durability.transaction.HistoryBuilder[source]

Ensures that the message history does not end up in a malformed state

classmethod agents_called() → frozenset[int][source]

Return type:: frozenset[int]

classmethod append_message(message: ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage) → None[source]

Append message to the ambient history, if it is legal where it lands.

Both checks are about position rather than content, which is why they sit here: every message the harness records passes through this method, including the ones a failed attempt records on its way out, and those are the ones that get a history into a shape no provider will accept.

Return type:: None

call_agent(skill, *args, **kwargs)[source]

Run the call in a transaction over the agent’s own history.

The buffer starts as a copy, so a call that raises leaves the agent’s history as it found it. write_back is keyed on the identity of that history: the outermost call for a given agent commits, while a nested call on the same agent – a tool invoking another of its skills – contributes to the same buffer instead of committing a second time. Being inside some other agent’s transaction does not count, which is what keeps a cross-agent nested call from being mistaken for a same-agent one.

call_assistant(*args, **kwargs)[source]

Record the assistant’s reply, including the replies that failed.

A decoding failure appends the raw message and the feedback describing what was wrong with it, because that pair is what the next attempt reads; the exception then propagates to whatever is retrying. Abandoned sibling tool calls are answered first (see _answer_abandoned_tool_calls) so the buffer stays well-formed enough to resend.

call_system(*args, **kwargs)[source]

Record the system message, but only as the first message of a history.

A history that already has content was restored or inherited, and its system message is already at position zero; appending a second one would leave two, which no provider accepts.

call_tool(*args, **kwargs)[source]

Record the tool result, including a failed call’s traceback.

Every advertised call must be answered, so a raising tool still appends a message before the exception continues outward: the feedback message is that answer, and leaving the call unanswered would make the conversation unresendable.

call_user(*args, **kwargs)[source]: Record the user message that opens a turn.

classmethod get_history() → MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage][source]

Return type:: MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage]

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_tool, (tool_call: effectful.handlers.llm.harness.serialization.DecodedToolCall[T]) -> ToolResult[T]): <function HistoryBuilder.call_tool>, Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function HistoryBuilder.call_assistant>, Operation(call_user, (user_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionUserMessage): <function HistoryBuilder.call_user>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function HistoryBuilder.call_system>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function HistoryBuilder.call_agent>}

effectful.handlers.llm.harness.durability.transaction.compact_(history: MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], tool_call_id: ToolCallID, scope: CompactionScope) → None[source]

Compact a history in-place, keeping the request and the asking round.

tool_call_id identifies the call that asked, and so the round to keep: the assistant message advertising it, and everything after (which is exactly the tool messages answering it and its siblings, whether they were appended before this one or are still to come – truncation only ever removes messages ahead of that assistant message, so no tool message is ever orphaned from the call it answers).

The request kept is the last user message before that round – the one this call opened – carried over untouched.

A no-op for scope="none", and whenever the shape this reads off the history is not the one it expects: no assistant message advertising tool_call_id, or no user message ahead of it. Declining is the right failure here; a compaction is a courtesy, and a wrong guess about the shape would corrupt the history the call still has to finish over. A conversation that opens with something other than a system message simply has no head to keep, which is not a failure.

Return type:: None

effectful.handlers.llm.harness.durability.transaction.transaction(prefix: MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage] | None = None, *, write_back: bool = True) → Generator[MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], None, None][source]

Context manager for a message transaction.

The buffer starts as a copy of prefix, and writing back reconciles the two. There are two ways a transaction can end, distinguished by whether the inherited prefix survived in the buffer:

Appended to. The usual case: the buffer still opens with the same message objects it was seeded with, so writing back means handing prefix the tail the transaction produced. The split point is taken on entry, so a prefix that grew by some other route meanwhile still receives exactly this transaction’s messages.
Rewritten. A compaction (see ~effectful.handlers.llm.harness.durability.compaction.compact) drops messages the transaction inherited, so the buffer no longer extends its seed – it may even be shorter than it. There is no tail to hand over then; the buffer is the new history, and appending buffer[start:] would leave prefix uncompacted and silently discard everything the transaction added. Adopt the buffer wholesale instead, keeping any concurrent growth after the split point, which is the same guarantee the appending case makes.

Return type:: Generator[MutableSequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], None, None]

class effectful.handlers.llm.harness.durability.retrying.TenacityRetryer(catch_tool_errors: type[BaseException] | tuple[type[BaseException], ...] = <class 'Exception'>, stop: ~tenacity.stop.stop_base = <tenacity.stop.stop_after_attempt object>, **kwargs)[source]

A reply that cannot be decoded is not the end of the attempt. If your answer or a tool call comes back malformed – wrong shape for the return type, a tool call whose arguments do not fit the signature – you are shown the decoding error and asked again, with the failed reply and the error visible in the conversation. A malformed tool call is reported as that call’s result; a malformed answer has no call to report against, so it comes back as a user message. That message is not a new question, and says so: it is the same call, asked again.

That budget is finite — a handful of attempts, fixed by whoever configured this harness — after which the error is raised to the caller and the call fails. So a second attempt should not resubmit the first one with cosmetic edits. If the same shape has already been rejected once, the error is telling you the shape is wrong; change it. None of these attempts leave a trace once one succeeds, so you will not see the failed exchanges again in later turns.

A tool that raises is different, and not a failure of this kind. The traceback comes back as that tool’s result and the conversation continues normally, without consuming a retry. Read it as data about the call you made – a wrong argument, a missing file – and make the next call.

Configure the retry policy.

Args:

catch_tool_errors: Exception type(s) to catch during tool execution.: Can be a single exception class or a tuple of exception classes. Defaults to Exception (catches all exceptions).
stop: tenacity stop condition for retrying call_assistant. Defaults: to tenacity.stop_after_attempt(4), which stops after 4 attempts.
**kwargs: Additional keyword arguments forwarded to: tenacity.Retrying.

call_assistant(messages: Sequence[ChatCompletionAssistantMessage | ChatCompletionToolMessage | ChatCompletionSystemMessage | ChatCompletionUserMessage], response_type: type[T], env: Mapping[str, Any], tools: Set[Tool] = frozenset({})) → AssistantResult[source]

Retry the request while the reply fails to decode.

ToolCallDecodingError and ResultDecodingError are the retryable failures: both mean the model replied but the reply could not be turned into the requested type. Anything else propagates on the first raise.

Each attempt re-reads buffer: the transaction makes it the ambient history for the duration, so HistoryBuilder appends the failed response and its error feedback there, and the next attempt sends them to the model. write_back=False then discards that scratch work, and only the response that finally succeeded joins the real history – which is why the caller never sees the malformed attempts.

Return type:: GenericAlias[TypeVar(T)]

call_assistant_retryer: Retrying

call_tool(tool_call: DecodedToolCall) → ToolResult[source]

Handle tool execution with runtime error capture.

Runtime errors from tool execution are captured and returned as error messages to the LLM. Only exceptions matching catch_tool_errors are caught; others propagate up.

A captured failure is reported as is_final=False, and the error object itself is returned in place of a result, so an enclosing handler can see that the call failed and decline to finalize on it – see effectful.handlers.llm.harness.synthesis.body.FinalBodySynthesizer. The completion loop therefore continues: the model sees the error message and gets another turn to retry.

Return type:: GenericAlias[TypeVar(T)]

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_tool, (tool_call: effectful.handlers.llm.harness.serialization.DecodedToolCall[T]) -> ToolResult[T]): <function TenacityRetryer.call_tool>, Operation(call_assistant, (messages: collections.abc.Sequence[litellm.types.llms.openai.ChatCompletionAssistantMessage | litellm.types.llms.openai.ChatCompletionToolMessage | litellm.types.llms.openai.ChatCompletionSystemMessage | litellm.types.llms.openai.ChatCompletionUserMessage], response_type: type[T], env: collections.abc.Mapping[str, typing.Any], tools: collections.abc.Set[effectful.handlers.llm.types.Tool] = frozenset()) -> AssistantResult[T]): <function TenacityRetryer.call_assistant>, Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>}

Checkpointing of Agent history and state to a SQLite database.

Install SQLitePersister alongside AgentLoop, LiteLLMConfigurer and HistoryBuilder:

with (
    handler(AgentLoop()),
    handler(LiteLLMConfigurer()),
    handler(HistoryBuilder()),
    handler(SQLitePersister(Path("./state/checkpoints.db"))),
):
    bot.ask("question")

HistoryBuilder is what opens the transaction a call’s messages accumulate in, so a stack without it has no history for this handler to checkpoint. ~effectful.handlers.llm.harness.harness assembles all of these; assemble them by hand only to leave one out.

It composes with ~effectful.handlers.llm.harness.durability.retrying.TenacityRetryer:

with (
    handler(AgentLoop()),
    handler(LiteLLMConfigurer()),
    handler(HistoryBuilder()),
    handler(TenacityRetryer()),
    handler(SQLitePersister(Path("./state/checkpoints.db"))),
):
    bot.ask("question")

There is deliberately no crash-recovery “handoff” note written on restore: the last successful checkpoint is already a complete, uncorrupted transcript, and the caller is expected to simply retry the request that didn’t finish.

class effectful.handlers.llm.harness.durability.persistence.SQLitePersister(db_path: Path) → None[source]

This conversation outlives the process. When a call you are answering returns, the whole exchange and the agent’s declared fields are written to disk, and the next time this agent runs – in a later process, days from now – they are restored. The history you are reading may therefore begin long before this run started.

So anything you set on the agent persists, and is worth setting deliberately: notes, accumulated findings, a running summary. Conversely, do not re-derive what an earlier turn already established and recorded; it is in front of you because it was saved, not because it was just computed.

A call that raises saves nothing. If you are heading toward an error, an intermediate result you want kept should be recorded before the failure, not after it.

Open (creating if absent) the checkpoint database.

WAL mode buys crash tolerance: if the process is killed mid-write, SQLite’s journal-based recovery keeps the database consistent. synchronous=NORMAL is the usual companion to WAL – it trades an fsync per commit for one per checkpoint, which is the right trade when the alternative to a lost final commit is re-running the call.

All state is read from and written to the database directly, with no in-memory cache to go stale, so several processes may share one file.

Args:: db_path: Path to the SQLite database file.

call_agent(skill: Skill[P, T], *args: P, **kwargs: P) → T[source]

Checkpoint the agent after the call returns.

The save happens after fwd, so nothing is written when the call raises: a Skill call’s work happens against a private copy of the agent’s history that is only written back on success (see AgentLoop.call_agent), so an interrupted call’s partial exchange – and any other in-process state not captured by __history__, such as a PythonRepl session – is unrecoverable regardless of what this handler does.

Two gates decide whether anything is written. The skill must be bound to an agent (__history__), and that agent must have been given an explicit agent_id (see Agent): a transient agent, the default, is never written to the database, even when nested inside a persisted agent’s call under this same handler.

Nested calls – a tool invoking another skill on the same agent – run this rule too, so each writes its own checkpoint on the way out. That is harmless rather than intended: the agent’s history is one shared object, so the enclosing call’s save overwrites the nested one with a superset, and the row left behind is the state as of the outermost return.

Return type:: TypeVar(T)

db_path: Path

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function PromptInjectingInterpretation.call_system>, Operation(_checkpoint_connection, () -> sqlite3.Connection | None): <function SQLitePersister._get_checkpoint_connection>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function SQLitePersister.call_agent>}

Observability

Handlers that expose what happened during a call without changing it.

class effectful.handlers.llm.harness.observability.rich.RichTerminalRenderer(console: Console = <factory>, _live_lock: lock = <factory>, _printed: list[str] = <factory>) → None[source]

Stream completion and live-render the message sequence.

Opt-in debugging handler: forces streaming so that reasoning, generation and tool-call arguments appear as they are produced, then reassembles a normal ModelResponse via litellm.stream_chunk_builder() so the rest of the pipeline is unchanged.

Each message is printed once, as a panel, in the order the conversation reaches it; only the turn currently being streamed lives inside a rich.live.Live region, and only that turn is redrawn. The alternative – rebuilding the whole history into the live region on every chunk – is what a terminal cannot do: rich erases a frame by rewinding the cursor over it, the rewind is clamped at the top of the screen, and a frame taller than the screen therefore accumulates one full copy of the conversation per refresh. Measured on a three-turn run of llm_examples/reasoning/countdown.py, that came to 1.2 MB and 7,208 lines of output carrying 335 distinct ones, with the system and user panels reprinted 79 times each.

completion(*args, **kwargs) → Any[source]

Stream and live-render this completion, or – if another already holds the terminal, or if the stream breaks – let it run unstreamed and print it as a settled panel.

A Live region owns the console for its duration, and there is one console. Concurrent skill calls (the asyncio.gather + asyncio.to_thread fan-out several examples demonstrate) would otherwise open overlapping Live regions and interleave two redraws into the same rows. Acquiring without blocking is what keeps that fan-out parallel: a caller that loses the race proceeds immediately down the settled path rather than queueing behind the live one.

The same fallback covers a stream that dies in flight. Streaming here is a debugging affordance this handler adds to a request that did not ask for it, so it also owns the cost: a long-lived streamed read shares a connection pool with whatever else the process is doing, and losing that race is not a reason to fail a call that would have succeeded unstreamed. The retry is safe because a broken stream yields no result to duplicate.

Return type:: Any

console: Console

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(completion, (model: str, messages: list = [], timeout: float | str | openai.Timeout | None = None, temperature: float | None = None, top_p: float | None = None, n: int | None = None, stream: bool | None = None, stream_options: dict | None = None, stop=None, max_completion_tokens: int | None = None, max_tokens: int | None = None, modalities: list[Literal['text', 'audio']] | None = None, prediction: openai.types.chat.chat_completion_prediction_content_param.ChatCompletionPredictionContentParam | None = None, audio: openai.types.chat.chat_completion_audio_param.ChatCompletionAudioParam | None = None, presence_penalty: float | None = None, frequency_penalty: float | None = None, logit_bias: dict | None = None, user: str | None = None, reasoning_effort: Literal['none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'default'] | None = None, verbosity: Literal['low', 'medium', 'high'] | None = None, response_format: dict | type[pydantic.main.BaseModel] | None = None, seed: int | None = None, tools: list | None = None, tool_choice: str | dict | None = None, logprobs: bool | None = None, top_logprobs: int | None = None, parallel_tool_calls: bool | None = None, web_search_options: litellm.types.llms.openai.OpenAIWebSearchOptions | None = None, include_server_side_tool_invocations: bool | None = None, deployment_id=None, extra_headers: dict | None = None, safety_identifier: str | None = None, service_tier: str | None = None, store: bool | None = None, prompt_cache_key: str | None = None, functions: list | None = None, function_call: str | None = None, base_url: str | None = None, api_version: str | None = None, api_key: str | None = None, model_list: list | None = None, thinking: litellm.types.llms.anthropic.AnthropicThinkingParam | None = None, shared_session: ForwardRef('ClientSession') | None = None, enable_json_schema_validation: bool | None = None, **kwargs) -> Any): <function RichTerminalRenderer.completion>}

class effectful.handlers.llm.harness.observability.dump.SystemPromptDumper(path: Path) → None[source]

Dump the system prompt produced by call_system to a Markdown file.

Opt-in debugging handler: intercepts call_system, forwards to let the prompt be assembled and installed as usual, then writes the resulting system message content to path, overwriting the whole file each time.

call_system(harness_prompt, agent_prompt)[source]

Write the assembled system message to path, then return it.

Forwards first, so what lands on disk is the finished prompt every other handler has contributed to, not this handler’s view of it. The file is overwritten each time.

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_system, (harness_prompt: effectful.handlers.llm.harness.serialization.PromptSection, agent_prompt: effectful.handlers.llm.harness.serialization.PromptSection) -> litellm.types.llms.openai.ChatCompletionSystemMessage): <function SystemPromptDumper.call_system>}

path: Path

class effectful.handlers.llm.harness.observability.langfuse.LangfuseTracer(client: Langfuse = <factory>) → None[source]

Traces Tool, Skill, and completion calls with Langfuse.

Compose with a provider via handler() to add tracing:

with handler(provider), handler(LangfuseTracer()):
    print(limerick(theme))

call_agent(skill: Skill, *args, **kwargs)[source]

Trace one Skill call as a Langfuse agent observation.

This is the observation the completions and tool calls of the call nest under, so a trace mirrors the call structure – including nested Skill calls, which appear as child agents.

call_tool(tool_call: DecodedToolCall)[source]

Trace one tool call as a Langfuse tool observation.

The arguments are encoded best-effort, falling back to repr: a call decoded from a Python expression binds real runtime values, which need not be encodable at all, and tracing must never be the thing that fails a call.

client: Langfuse

completion(*args, **kwargs)[source]

Trace one model request as a Langfuse generation observation.

Request parameters, the resolved model and the token usage are recorded alongside the messages, since a trace whose cost cannot be attributed is most of the reason to have one. The tools go in the input only when the request carried any, keeping a plain completion’s trace readable.

implementations: dict[Operation[..., T], Callable[[...], V]] = {Operation(call_tool, (tool_call: effectful.handlers.llm.harness.serialization.DecodedToolCall[T]) -> ToolResult[T]): <function LangfuseTracer.call_tool>, ApplyOperation(__apply__, (op: effectful.ops.types.Operation[A, B], *args: A.args, **kwargs: A.kwargs) -> B): <function LangfuseTracer.call_agent>, Operation(completion, (model: str, messages: list = [], timeout: float | str | openai.Timeout | None = None, temperature: float | None = None, top_p: float | None = None, n: int | None = None, stream: bool | None = None, stream_options: dict | None = None, stop=None, max_completion_tokens: int | None = None, max_tokens: int | None = None, modalities: list[Literal['text', 'audio']] | None = None, prediction: openai.types.chat.chat_completion_prediction_content_param.ChatCompletionPredictionContentParam | None = None, audio: openai.types.chat.chat_completion_audio_param.ChatCompletionAudioParam | None = None, presence_penalty: float | None = None, frequency_penalty: float | None = None, logit_bias: dict | None = None, user: str | None = None, reasoning_effort: Literal['none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'default'] | None = None, verbosity: Literal['low', 'medium', 'high'] | None = None, response_format: dict | type[pydantic.main.BaseModel] | None = None, seed: int | None = None, tools: list | None = None, tool_choice: str | dict | None = None, logprobs: bool | None = None, top_logprobs: int | None = None, parallel_tool_calls: bool | None = None, web_search_options: litellm.types.llms.openai.OpenAIWebSearchOptions | None = None, include_server_side_tool_invocations: bool | None = None, deployment_id=None, extra_headers: dict | None = None, safety_identifier: str | None = None, service_tier: str | None = None, store: bool | None = None, prompt_cache_key: str | None = None, functions: list | None = None, function_call: str | None = None, base_url: str | None = None, api_version: str | None = None, api_key: str | None = None, model_list: list | None = None, thinking: litellm.types.llms.anthropic.AnthropicThinkingParam | None = None, shared_session: ForwardRef('ClientSession') | None = None, enable_json_schema_validation: bool | None = None, **kwargs) -> Any): <function LangfuseTracer.completion>}

Jax

effectful.handlers.jax.bind_dims(value: Scoped(ordinal=frozenset({B, A}))], *names: Scoped(ordinal=frozenset({B}))]) → Scoped(ordinal=frozenset({A}))][source]

Convert named dimensions to positional dimensions.

Parameters:

t – An array.
args – Named dimensions to convert to positional dimensions. These positional dimensions will appear at the beginning of the shape.

Return type:

TypeVar(T)

Returns:

An array with the named dimensions in args converted to positional dimensions.

Example usage:

>>> import jax.numpy as jnp
>>> from effectful.ops.syntax import defop
>>> a, b = defop(jax.Array, name='a'), defop(jax.Array, name='b')
>>> t = jax_getitem(jnp.ones((2, 3)), [a(), b()])
>>> bind_dims(t, b, a).shape
(3, 2)

effectful.handlers.jax.jax_getitem(*args, **kwargs) → Array[source]

Operation for indexing an array. Unlike the standard __getitem__ method, this operation correctly handles indexing with terms.

Return type:: Array

effectful.handlers.jax.jit(f, *args, **kwargs)[source]

effectful.handlers.jax.sizesof(term: Expr) → Mapping[Operation[(), Array], int][source]

Return the sizes of named dimensions in an array expression.

Sizes are inferred from the array shape.

Parameters:: value – An array expression.
Return type:: Mapping[Operation[(), Array], int]
Returns:: A mapping from named dimensions to their sizes.

Example usage:

>>> a, b = defop(jax.Array, name='a'), defop(jax.Array, name='b')
>>> sizes = sizesof(jax_getitem(jnp.ones((2, 3)), [a(), b()]))
>>> assert sizes[a] == 2 and sizes[b] == 3

effectful.handlers.jax.unbind_dims(value: Scoped(ordinal=frozenset({B, A}))], *names: Scoped(ordinal=frozenset({B}))]) → Scoped(ordinal=frozenset({B, A}))][source]

Convert positional dimensions to named dimensions.

Return type:: TypeVar(T)

Numpyro

effectful.handlers.numpyro.BernoulliLogits(logits, **kwargs) → BernoulliLogits[source]

Return type:: BernoulliLogits

class effectful.handlers.numpyro.BernoulliLogitsTerm(ty, op, logits, **kwargs)[source]

effectful.handlers.numpyro.BernoulliProbs(probs, **kwargs) → BernoulliProbs[source]

Return type:: BernoulliProbs

class effectful.handlers.numpyro.BernoulliProbsTerm(ty, op, probs, **kwargs)[source]

effectful.handlers.numpyro.Beta(concentration1, concentration0, **kwargs) → Beta[source]

Return type:: Beta

class effectful.handlers.numpyro.BetaTerm(ty, op, concentration1, concentration0, **kwargs)[source]

effectful.handlers.numpyro.BinomialLogits(logits, total_count=1, **kwargs) → BinomialLogits[source]

Return type:: BinomialLogits

class effectful.handlers.numpyro.BinomialLogitsTerm(ty, op, logits, total_count, **kwargs)[source]

effectful.handlers.numpyro.BinomialProbs(probs, total_count=1, **kwargs) → BinomialProbs[source]

Return type:: BinomialProbs

class effectful.handlers.numpyro.BinomialProbsTerm(ty, op, probs, total_count, **kwargs)[source]

effectful.handlers.numpyro.CategoricalLogits(logits, **kwargs) → CategoricalLogits[source]

Return type:: CategoricalLogits

class effectful.handlers.numpyro.CategoricalLogitsTerm(ty, op, logits, **kwargs)[source]

effectful.handlers.numpyro.CategoricalProbs(probs, **kwargs) → CategoricalProbs[source]

Return type:: CategoricalProbs

class effectful.handlers.numpyro.CategoricalProbsTerm(ty, op, probs, **kwargs)[source]

effectful.handlers.numpyro.Cauchy(loc=0.0, scale=1.0, **kwargs) → Cauchy[source]

Return type:: Cauchy

class effectful.handlers.numpyro.CauchyTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.Chi2(df, **kwargs) → Chi2[source]

Return type:: Chi2

class effectful.handlers.numpyro.Chi2Term(ty, op, df, **kwargs)[source]

effectful.handlers.numpyro.Delta(v=0.0, log_density=0.0, event_dim=0, **kwargs) → Delta[source]

Return type:: Delta

class effectful.handlers.numpyro.DeltaTerm(ty, op, v, log_density, event_dim, **kwargs)[source]

effectful.handlers.numpyro.Dirichlet(concentration, **kwargs) → Dirichlet[source]

Return type:: Dirichlet

effectful.handlers.numpyro.DirichletMultinomial(concentration, total_count=1, **kwargs) → DirichletMultinomial[source]

Return type:: DirichletMultinomial

class effectful.handlers.numpyro.DirichletMultinomialTerm(ty, op, concentration, total_count, **kwargs)[source]

class effectful.handlers.numpyro.DirichletTerm(ty, op, concentration, **kwargs)[source]

effectful.handlers.numpyro.Exponential(rate=1.0, **kwargs) → Exponential[source]

Return type:: Exponential

class effectful.handlers.numpyro.ExponentialTerm(ty, op, rate, **kwargs)[source]

effectful.handlers.numpyro.Gamma(concentration, rate=1.0, **kwargs) → Gamma[source]

Return type:: Gamma

class effectful.handlers.numpyro.GammaTerm(ty, op, concentration, rate, **kwargs)[source]

effectful.handlers.numpyro.GeometricLogits(logits, **kwargs) → GeometricLogits[source]

Return type:: GeometricLogits

class effectful.handlers.numpyro.GeometricLogitsTerm(ty, op, logits, **kwargs)[source]

effectful.handlers.numpyro.GeometricProbs(probs, **kwargs) → GeometricProbs[source]

Return type:: GeometricProbs

class effectful.handlers.numpyro.GeometricProbsTerm(ty, op, probs, **kwargs)[source]

effectful.handlers.numpyro.Gumbel(loc=0.0, scale=1.0, **kwargs) → Gumbel[source]

Return type:: Gumbel

class effectful.handlers.numpyro.GumbelTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.HalfCauchy(scale=1.0, **kwargs) → HalfCauchy[source]

Return type:: HalfCauchy

class effectful.handlers.numpyro.HalfCauchyTerm(ty, op, scale, **kwargs)[source]

effectful.handlers.numpyro.HalfNormal(scale=1.0, **kwargs) → HalfNormal[source]

Return type:: HalfNormal

class effectful.handlers.numpyro.HalfNormalTerm(ty, op, scale, **kwargs)[source]

effectful.handlers.numpyro.Independent(base_dist, reinterpreted_batch_ndims, **kwargs) → Independent[source]

Return type:: Independent

class effectful.handlers.numpyro.IndependentTerm(ty, op, base_dist, reinterpreted_batch_ndims, **kwargs)[source]

effectful.handlers.numpyro.Kumaraswamy(concentration1, concentration0, **kwargs) → Kumaraswamy[source]

Return type:: Kumaraswamy

class effectful.handlers.numpyro.KumaraswamyTerm(ty, op, concentration1, concentration0, **kwargs)[source]

effectful.handlers.numpyro.LKJCholesky(dim, concentration=1.0, **kwargs) → LKJCholesky[source]

Return type:: LKJCholesky

class effectful.handlers.numpyro.LKJCholeskyTerm(ty, op, dim, concentration, **kwargs)[source]

effectful.handlers.numpyro.Laplace(loc=0.0, scale=1.0, **kwargs) → Laplace[source]

Return type:: Laplace

class effectful.handlers.numpyro.LaplaceTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.LogNormal(loc=0.0, scale=1.0, **kwargs) → LogNormal[source]

Return type:: LogNormal

class effectful.handlers.numpyro.LogNormalTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.Logistic(loc=0.0, scale=1.0, **kwargs) → Logistic[source]

Return type:: Logistic

class effectful.handlers.numpyro.LogisticTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.LowRankMultivariateNormal(loc, cov_factor, cov_diag, **kwargs) → LowRankMultivariateNormal[source]

Return type:: LowRankMultivariateNormal

class effectful.handlers.numpyro.LowRankMultivariateNormalTerm(ty, op, loc, cov_factor, cov_diag, **kwargs)[source]

effectful.handlers.numpyro.MultinomialLogits(logits, total_count=1, **kwargs) → MultinomialLogits[source]

Return type:: MultinomialLogits

class effectful.handlers.numpyro.MultinomialLogitsTerm(ty, op, logits, total_count, **kwargs)[source]

effectful.handlers.numpyro.MultinomialProbs(probs, total_count=1, **kwargs) → MultinomialProbs[source]

Return type:: MultinomialProbs

class effectful.handlers.numpyro.MultinomialProbsTerm(ty, op, probs, total_count, **kwargs)[source]

effectful.handlers.numpyro.MultivariateNormal(loc=0.0, covariance_matrix=None, precision_matrix=None, scale_tril=None, **kwargs) → MultivariateNormal[source]

Return type:: MultivariateNormal

class effectful.handlers.numpyro.MultivariateNormalTerm(ty, op, loc, covariance_matrix, precision_matrix, scale_tril, **kwargs)[source]

class effectful.handlers.numpyro.Naming(name_to_dim: Mapping[Operation[(), Array], int])[source]

A mapping from dimensions (indexed from the right) to names.

apply(value: Array) → Array[source]

Return type:: Array

static from_shape(names: Collection[Operation[(), Array]], event_dims: int) → Naming[source]

Create a naming from a set of indices and the number of event dimensions.

Return type:: Naming

effectful.handlers.numpyro.NegativeBinomialLogits(total_count, logits, **kwargs) → NegativeBinomialLogits[source]

Return type:: NegativeBinomialLogits

class effectful.handlers.numpyro.NegativeBinomialLogitsTerm(ty, op, total_count, logits, **kwargs)[source]

effectful.handlers.numpyro.NegativeBinomialProbs(total_count, probs, **kwargs) → NegativeBinomialProbs[source]

Return type:: NegativeBinomialProbs

class effectful.handlers.numpyro.NegativeBinomialProbsTerm(ty, op, total_count, probs, **kwargs)[source]

effectful.handlers.numpyro.Normal(loc=0.0, scale=1.0, **kwargs) → Normal[source]

Return type:: Normal

class effectful.handlers.numpyro.NormalTerm(ty, op, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.Pareto(scale, alpha, **kwargs) → Pareto[source]

Return type:: Pareto

class effectful.handlers.numpyro.ParetoTerm(ty, op, scale, alpha, **kwargs)[source]

effectful.handlers.numpyro.Poisson(rate, **kwargs) → Poisson[source]

Return type:: Poisson

class effectful.handlers.numpyro.PoissonTerm(ty, op, rate, **kwargs)[source]

effectful.handlers.numpyro.RelaxedBernoulliLogits(temperature, logits, **kwargs) → RelaxedBernoulliLogits[source]

Return type:: RelaxedBernoulliLogits

class effectful.handlers.numpyro.RelaxedBernoulliLogitsTerm(ty, op, temperature, logits, **kwargs)[source]

effectful.handlers.numpyro.StudentT(df, loc=0.0, scale=1.0, **kwargs) → StudentT[source]

Return type:: StudentT

class effectful.handlers.numpyro.StudentTTerm(ty, op, df, loc, scale, **kwargs)[source]

effectful.handlers.numpyro.Uniform(low=0.0, high=1.0, **kwargs) → Uniform[source]

Return type:: Uniform

class effectful.handlers.numpyro.UniformTerm(ty, op, low, high, **kwargs)[source]

effectful.handlers.numpyro.VonMises(loc, concentration, **kwargs) → VonMises[source]

Return type:: VonMises

class effectful.handlers.numpyro.VonMisesTerm(ty, op, loc, concentration, **kwargs)[source]

effectful.handlers.numpyro.Weibull(scale, concentration, **kwargs) → Weibull[source]

Return type:: Weibull

class effectful.handlers.numpyro.WeibullTerm(ty, op, scale, concentration, **kwargs)[source]

effectful.handlers.numpyro.Wishart(df, scale_tril, **kwargs) → Wishart[source]

Return type:: Wishart

class effectful.handlers.numpyro.WishartTerm(ty, op, df, scale_tril, **kwargs)[source]

effectful.handlers.numpyro.entropy(self) → Array

Return type:: Array

effectful.handlers.numpyro.enumerate_support(self, expand=True) → Array

Return type:: Array

effectful.handlers.numpyro.expand(self, batch_shape) → Distribution

Return type:: Distribution

effectful.handlers.numpyro.expand_to_batch_shape(tensor, batch_ndims, expanded_batch_shape)[source]

Expands a tensor of shape batch_shape + remaining_shape to expanded_batch_shape + remaining_shape.

Args:: tensor: JAX array with shape batch_shape + event_shape expanded_batch_shape: tuple of the desired expanded batch dimensions event_ndims: number of dimensions in the event_shape
Returns:: A JAX array with shape expanded_batch_shape + event_shape

effectful.handlers.numpyro.log_prob(self, value) → Array

Return type:: Array

effectful.handlers.numpyro.rsample(self, key, sample_shape=()) → Array

Return type:: Array

effectful.handlers.numpyro.sample(self, key, sample_shape=()) → Array

Return type:: Array

effectful.handlers.numpyro.to_event(self, reinterpreted_batch_ndims=None) → Distribution

Return type:: Distribution

Pyro

class effectful.handlers.pyro.Naming(name_to_dim: Mapping[Operation[(), Tensor], int])[source]

A mapping from dimensions (indexed from the right) to names.

apply(value: Tensor) → Tensor[source]

Return type:: Tensor

static from_shape(names: Collection[Operation[(), Tensor]], event_dims: int) → Naming[source]

Create a naming from a set of indices and the number of event dimensions.

Return type:: Naming

class effectful.handlers.pyro.PyroShim[source]

Pyro handler that wraps all sample sites in a custom effectful type.

Note

This handler should be installed around any Pyro model that you want to use effectful handlers with.

Example usage:

>>> import pyro.distributions as dist
>>> from effectful.ops.semantics import fwd, handler
>>> torch.distributions.Distribution.set_default_validate_args(False)

It can be used as a decorator:

>>> @PyroShim()
... def model():
...     return pyro.sample("x", dist.Normal(0, 1))

It can also be used as a context manager:

>>> with PyroShim():
...     x = pyro.sample("x", dist.Normal(0, 1))

When PyroShim is installed, all sample sites perform the pyro_sample() effect, which can be handled by an effectful interpretation.

>>> def log_sample(name, *args, **kwargs):
...     print(f"Sampled {name}")
...     return fwd()

>>> with PyroShim(), handler({pyro_sample: log_sample}):
...     x = pyro.sample("x", dist.Normal(0, 1))
...     y = pyro.sample("y", dist.Normal(0, 1))
Sampled x
Sampled y

effectful.handlers.pyro.pyro_module_shim(module: type[PyroModule]) → type[PyroModule][source]

Wrap a PyroModule in a PyroShim.

Returns a new subclass of PyroModule that wraps calls to forward() in a PyroShim.

Example usage:

class SimpleModel(PyroModule):
    def forward(self):
        return pyro.sample("y", dist.Normal(0, 1))

SimpleModelShim = pyro_module_shim(SimpleModel)

Return type:: type[PyroModule]

effectful.handlers.pyro.pyro_sample(name: str, fn: TorchDistributionMixin, *args, obs: Tensor | None = None, obs_mask: BoolTensor | None = None, mask: BoolTensor | None = None, infer: InferDict | None = None, **kwargs) → Tensor[source]

Operation to sample from a Pyro distribution. See pyro.sample().

Return type:: Tensor

Torch

effectful.handlers.torch.grad(func: Callable[[_P], Any], argnums: int | tuple[int, ...] = 0, has_aux: bool = False) → Callable[[_P], Any]

grad operator helps computing gradients of func with respect to the input(s) specified by argnums. This operator can be nested to compute higher-order gradients.

Return type:: Callable[[ParamSpec(_P, bound= None)], Any]

Args:

func (Callable): A Python function that takes one or more arguments.: Must return a single-element Tensor. If specified has_aux equals True, function can return a tuple of single-element Tensor and other auxiliary objects: (output, aux).
argnums (int or Tuple[int]): Specifies arguments to compute gradients with respect to.: argnums can be single integer or tuple of integers. Default: 0.
has_aux (bool): Flag indicating that func returns a tensor and other: auxiliary objects: (output, aux). Default: False.

Returns:

Function to compute gradients with respect to its inputs. By default, the output of the function is the gradient tensor(s) with respect to the first argument. If specified has_aux equals True, tuple of gradients and output auxiliary objects is returned. If argnums is a tuple of integers, a tuple of output gradients with respect to each argnums value is returned.

Example of using grad:

>>> # xdoctest: +SKIP
>>> from torch.func import grad
>>> x = torch.randn([])
>>> cos_x = grad(lambda x: torch.sin(x))(x)
>>> assert torch.allclose(cos_x, x.cos())
>>>
>>> # Second-order gradients
>>> neg_sin_x = grad(grad(lambda x: torch.sin(x)))(x)
>>> assert torch.allclose(neg_sin_x, -x.sin())

When composed with vmap, grad can be used to compute per-sample-gradients:

>>> # xdoctest: +SKIP
>>> from torch.func import grad, vmap
>>> batch_size, feature_size = 3, 5
>>>
>>> def model(weights, feature_vec):
>>> # Very simple linear model with activation
>>>     assert feature_vec.dim() == 1
>>>     return feature_vec.dot(weights).relu()
>>>
>>> def compute_loss(weights, example, target):
>>>     y = model(weights, example)
>>>     return ((y - target) ** 2).mean()  # MSELoss
>>>
>>> weights = torch.randn(feature_size, requires_grad=True)
>>> examples = torch.randn(batch_size, feature_size)
>>> targets = torch.randn(batch_size)
>>> inputs = (weights, examples, targets)
>>> grad_weight_per_example = vmap(grad(compute_loss), in_dims=(None, 0, 0))(
...     *inputs
... )

Example of using grad with has_aux and argnums:

>>> # xdoctest: +SKIP
>>> from torch.func import grad
>>> def my_loss_func(y, y_pred):
>>>    loss_per_sample = (0.5 * y_pred - y) ** 2
>>>    loss = loss_per_sample.mean()
>>>    return loss, (y_pred, loss_per_sample)
>>>
>>> fn = grad(my_loss_func, argnums=(0, 1), has_aux=True)
>>> y_true = torch.rand(4)
>>> y_preds = torch.rand(4, requires_grad=True)
>>> out = fn(y_true, y_preds)
>>> # > output is ((grads w.r.t y_true, grads w.r.t y_preds), (y_pred, loss_per_sample))

Note

Using PyTorch torch.no_grad together with grad.

Case 1: Using torch.no_grad inside a function:

>>> # xdoctest: +SKIP
>>> def f(x):
>>>     with torch.no_grad():
>>>         c = x ** 2
>>>     return x - c

In this case, grad(f)(x) will respect the inner torch.no_grad.

Case 2: Using grad inside torch.no_grad context manager:

>>> # xdoctest: +SKIP
>>> with torch.no_grad():
>>>     grad(f)(x)

In this case, grad will respect the inner torch.no_grad, but not the outer one. This is because grad is a “function transform”: its result should not depend on the result of a context manager outside of f.

effectful.handlers.torch.jacfwd(func: Callable[..., Any], argnums: argnums_t = 0, has_aux: bool = False, *, randomness: str = 'error') → Callable[..., Any]

Computes the Jacobian of func with respect to the arg(s) at index argnum using forward-mode autodiff

Return type:: Callable[…, Any]

Args:

func (function): A Python function that takes one or more arguments,: one of which must be a Tensor, and returns one or more Tensors
argnums (int or tuple[int, …]): Optional, integer or tuple of integers,: saying which arguments to get the Jacobian with respect to. Default: 0.
has_aux (bool): Flag indicating that func returns a: (output, aux) tuple where the first element is the output of the function to be differentiated and the second element is auxiliary objects that will not be differentiated. Default: False.
randomness(str): Flag indicating what type of randomness to use.: See vmap() for more detail. Allowed: “different”, “same”, “error”. Default: “error”

Returns:

Returns a function that takes in the same inputs as func and returns the Jacobian of func with respect to the arg(s) at argnums. If has_aux is True, then the returned function instead returns a (jacobian, aux) tuple where jacobian is the Jacobian and aux is auxiliary objects returned by func.

Note

You may see this API error out with “forward-mode AD not implemented for operator X”. If so, please file a bug report and we will prioritize it. An alternative is to use jacrev(), which has better operator coverage.

A basic usage with a pointwise, unary operation will give a diagonal array as the Jacobian

>>> from torch.func import jacfwd
>>> x = torch.randn(5)
>>> jacobian = jacfwd(torch.sin)(x)
>>> expected = torch.diag(torch.cos(x))
>>> assert torch.allclose(jacobian, expected)

jacfwd() can be composed with vmap to produce batched Jacobians:

>>> from torch.func import jacfwd, vmap
>>> x = torch.randn(64, 5)
>>> jacobian = vmap(jacfwd(torch.sin))(x)
>>> assert jacobian.shape == (64, 5, 5)

If you would like to compute the output of the function as well as the jacobian of the function, use the has_aux flag to return the output as an auxiliary object:

>>> from torch.func import jacfwd
>>> x = torch.randn(5)
>>>
>>> def f(x):
>>>   return x.sin()
>>>
>>> def g(x):
>>>   result = f(x)
>>>   return result, result
>>>
>>> jacobian_f, f_x = jacfwd(g, has_aux=True)(x)
>>> assert torch.allclose(f_x, f(x))

Additionally, jacrev() can be composed with itself or jacrev() to produce Hessians

>>> from torch.func import jacfwd, jacrev
>>> def f(x):
>>>   return x.sin().sum()
>>>
>>> x = torch.randn(5)
>>> hessian = jacfwd(jacrev(f))(x)
>>> assert torch.allclose(hessian, torch.diag(-x.sin()))

By default, jacfwd() computes the Jacobian with respect to the first input. However, it can compute the Jacobian with respect to a different argument by using argnums:

>>> from torch.func import jacfwd
>>> def f(x, y):
>>>   return x + y ** 2
>>>
>>> x, y = torch.randn(5), torch.randn(5)
>>> jacobian = jacfwd(f, argnums=1)(x, y)
>>> expected = torch.diag(2 * y)
>>> assert torch.allclose(jacobian, expected)

Additionally, passing a tuple to argnums will compute the Jacobian with respect to multiple arguments

>>> from torch.func import jacfwd
>>> def f(x, y):
>>>   return x + y ** 2
>>>
>>> x, y = torch.randn(5), torch.randn(5)
>>> jacobian = jacfwd(f, argnums=(0, 1))(x, y)
>>> expectedX = torch.diag(torch.ones_like(x))
>>> expectedY = torch.diag(2 * y)
>>> assert torch.allclose(jacobian[0], expectedX)
>>> assert torch.allclose(jacobian[1], expectedY)

effectful.handlers.torch.jacrev(func: Callable[..., Any], argnums: int | tuple[int, ...] = 0, *, has_aux: bool = False, chunk_size: int | None = None, _preallocate_and_copy: bool = False) → Callable[..., Any]

Computes the Jacobian of func with respect to the arg(s) at index argnum using reverse mode autodiff

Note

Using chunk_size=1 is equivalent to computing the jacobian row-by-row with a for-loop i.e. the constraints of vmap() are not applicable.

Args:

func (function): A Python function that takes one or more arguments,: one of which must be a Tensor, and returns one or more Tensors
argnums (int or tuple[int, …]): Optional, integer or tuple of integers,: saying which arguments to get the Jacobian with respect to. Default: 0.
has_aux (bool): Flag indicating that func returns a: (output, aux) tuple where the first element is the output of the function to be differentiated and the second element is auxiliary objects that will not be differentiated. Default: False.
chunk_size (None or int): If None (default), use the maximum chunk size: (equivalent to doing a single vmap over vjp to compute the jacobian). If 1, then compute the jacobian row-by-row with a for-loop. If not None, then compute the jacobian chunk_size rows at a time (equivalent to doing multiple vmap over vjp). If you run into memory issues computing the jacobian, please try to specify a non-None chunk_size.

Returns:

Returns a function that takes in the same inputs as func and returns the Jacobian of func with respect to the arg(s) at argnums. If has_aux is True, then the returned function instead returns a (jacobian, aux) tuple where jacobian is the Jacobian and aux is auxiliary objects returned by func.

A basic usage with a pointwise, unary operation will give a diagonal array as the Jacobian

>>> from torch.func import jacrev
>>> x = torch.randn(5)
>>> jacobian = jacrev(torch.sin)(x)
>>> expected = torch.diag(torch.cos(x))
>>> assert torch.allclose(jacobian, expected)

If you would like to compute the output of the function as well as the jacobian of the function, use the has_aux flag to return the output as an auxiliary object:

>>> from torch.func import jacrev
>>> x = torch.randn(5)
>>>
>>> def f(x):
>>>   return x.sin()
>>>
>>> def g(x):
>>>   result = f(x)
>>>   return result, result
>>>
>>> jacobian_f, f_x = jacrev(g, has_aux=True)(x)
>>> assert torch.allclose(f_x, f(x))

jacrev() can be composed with vmap to produce batched Jacobians:

>>> from torch.func import jacrev, vmap
>>> x = torch.randn(64, 5)
>>> jacobian = vmap(jacrev(torch.sin))(x)
>>> assert jacobian.shape == (64, 5, 5)

Additionally, jacrev() can be composed with itself to produce Hessians

>>> from torch.func import jacrev
>>> def f(x):
>>>   return x.sin().sum()
>>>
>>> x = torch.randn(5)
>>> hessian = jacrev(jacrev(f))(x)
>>> assert torch.allclose(hessian, torch.diag(-x.sin()))

By default, jacrev() computes the Jacobian with respect to the first input. However, it can compute the Jacobian with respect to a different argument by using argnums:

>>> from torch.func import jacrev
>>> def f(x, y):
>>>   return x + y ** 2
>>>
>>> x, y = torch.randn(5), torch.randn(5)
>>> jacobian = jacrev(f, argnums=1)(x, y)
>>> expected = torch.diag(2 * y)
>>> assert torch.allclose(jacobian, expected)

Additionally, passing a tuple to argnums will compute the Jacobian with respect to multiple arguments

>>> from torch.func import jacrev
>>> def f(x, y):
>>>   return x + y ** 2
>>>
>>> x, y = torch.randn(5), torch.randn(5)
>>> jacobian = jacrev(f, argnums=(0, 1))(x, y)
>>> expectedX = torch.diag(torch.ones_like(x))
>>> expectedY = torch.diag(2 * y)
>>> assert torch.allclose(jacobian[0], expectedX)
>>> assert torch.allclose(jacobian[1], expectedY)

Note

Using PyTorch torch.no_grad together with jacrev. Case 1: Using torch.no_grad inside a function:

>>> def f(x):
>>>     with torch.no_grad():
>>>         c = x ** 2
>>>     return x - c

In this case, jacrev(f)(x) will respect the inner torch.no_grad.

Case 2: Using jacrev inside torch.no_grad context manager:

>>> with torch.no_grad():
>>>     jacrev(f)(x)

In this case, jacrev will respect the inner torch.no_grad, but not the outer one. This is because jacrev is a “function transform”: its result should not depend on the result of a context manager outside of f.

Return type:: Callable[…, Any]

effectful.handlers.torch.hessian(func: Callable[..., Any], argnums: argnums_t = 0) → Callable[..., Any]

Computes the Hessian of func with respect to the arg(s) at index argnum via a forward-over-reverse strategy.

The forward-over-reverse strategy (composing jacfwd(jacrev(func))) is a good default for good performance. It is possible to compute Hessians through other compositions of jacfwd() and jacrev() like jacfwd(jacfwd(func)) or jacrev(jacrev(func)).

Return type:: Callable[…, Any]

Args:

func (function): A Python function that takes one or more arguments,: one of which must be a Tensor, and returns one or more Tensors
argnums (int or tuple[int, …]): Optional, integer or tuple of integers,: saying which arguments to get the Hessian with respect to. Default: 0.

Returns:

Returns a function that takes in the same inputs as func and returns the Hessian of func with respect to the arg(s) at argnums.

Note

You may see this API error out with “forward-mode AD not implemented for operator X”. If so, please file a bug report and we will prioritize it. An alternative is to use jacrev(jacrev(func)), which has better operator coverage.

A basic usage with a R^N -> R^1 function gives a N x N Hessian:

>>> from torch.func import hessian
>>> def f(x):
>>>   return x.sin().sum()
>>>
>>> x = torch.randn(5)
>>> hess = hessian(f)(x)  # equivalent to jacfwd(jacrev(f))(x)
>>> assert torch.allclose(hess, torch.diag(-x.sin()))

effectful.handlers.torch.jvp(func: Callable[..., Any], primals: Any, tangents: Any, *, strict: bool = False, has_aux: bool = False) → tuple[Any, Any] | tuple[Any, Any, Any]

Standing for the Jacobian-vector product, returns a tuple containing the output of func(*primals) and the “Jacobian of func evaluated at primals” times tangents. This is also known as forward-mode autodiff.

Return type:: tuple[Any, Any] | tuple[Any, Any, Any]

Args:

func (function): A Python function that takes one or more arguments,: one of which must be a Tensor, and returns one or more Tensors
primals (Tensors): Positional arguments to func that must all be: Tensors. The returned function will also be computing the derivative with respect to these arguments
tangents (Tensors): The “vector” for which Jacobian-vector-product is: computed. Must be the same structure and sizes as the inputs to func.
has_aux (bool): Flag indicating that func returns a: (output, aux) tuple where the first element is the output of the function to be differentiated and the second element is other auxiliary objects that will not be differentiated. Default: False.

Returns:

Returns a (output, jvp_out) tuple containing the output of func evaluated at primals and the Jacobian-vector product. If has_aux is True, then instead returns a (output, jvp_out, aux) tuple.

Note

You may see this API error out with “forward-mode AD not implemented for operator X”. If so, please file a bug report and we will prioritize it.

jvp is useful when you wish to compute gradients of a function R^1 -> R^N

>>> from torch.func import jvp
>>> x = torch.randn([])
>>> f = lambda x: x * torch.tensor([1.0, 2.0, 3])
>>> warnings.filterwarnings(
...     "ignore", message=".*torch.jit.script"
... )  # docs: hide
>>> value, grad = jvp(f, (x,), (torch.tensor(1.0),))
>>> assert torch.allclose(value, f(x))
>>> assert torch.allclose(grad, torch.tensor([1.0, 2, 3]))

jvp() can support functions with multiple inputs by passing in the tangents for each of the inputs

>>> from torch.func import jvp
>>> x = torch.randn(5)
>>> y = torch.randn(5)
>>> f = lambda x, y: (x * y)
>>> _, output = jvp(f, (x, y), (torch.ones(5), torch.ones(5)))
>>> assert torch.allclose(output, x + y)

effectful.handlers.torch.vjp(func: Callable[..., Any], *primals: Any, has_aux: bool = False) → tuple[Any, Callable[..., Any]] | tuple[Any, Callable[..., Any], Any]

Standing for the vector-Jacobian product, returns a tuple containing the results of func applied to primals and a function that, when given cotangents, computes the reverse-mode Jacobian of func with respect to primals times cotangents.

Return type:: tuple[Any, Callable[…, Any]] | tuple[Any, Callable[…, Any], Any]

Args:

func (Callable[…, Any]): A Python function that takes one or more arguments. Must: return one or more Tensors.
primals (Tensors): Positional arguments to func that must all be: Tensors. The returned function will also be computing the derivative with respect to these arguments
has_aux (bool): Flag indicating that func returns a: (output, aux) tuple where the first element is the output of the function to be differentiated and the second element is other auxiliary objects that will not be differentiated. Default: False.

Returns:

Returns a (output, vjp_fn) tuple containing the output of func applied to primals and a function that computes the vjp of func with respect to all primals using the cotangents passed to the returned function. If has_aux is True, then instead returns a (output, vjp_fn, aux) tuple. The returned vjp_fn function will return a tuple of each VJP.

When used in simple cases, vjp() behaves the same as grad()

>>> x = torch.randn([5])
>>> f = lambda x: x.sin().sum()
>>> (_, vjpfunc) = torch.func.vjp(f, x)
>>> grad = vjpfunc(torch.tensor(1.0))[0]
>>> assert torch.allclose(grad, torch.func.grad(f)(x))

However, vjp() can support functions with multiple outputs by passing in the cotangents for each of the outputs

>>> x = torch.randn([5])
>>> f = lambda x: (x.sin(), x.cos())
>>> (_, vjpfunc) = torch.func.vjp(f, x)
>>> vjps = vjpfunc((torch.ones([5]), torch.ones([5])))
>>> assert torch.allclose(vjps[0], x.cos() + -x.sin())

vjp() can even support outputs being Python structs

>>> x = torch.randn([5])
>>> f = lambda x: {"first": x.sin(), "second": x.cos()}
>>> (_, vjpfunc) = torch.func.vjp(f, x)
>>> cotangents = {"first": torch.ones([5]), "second": torch.ones([5])}
>>> vjps = vjpfunc(cotangents)
>>> assert torch.allclose(vjps[0], x.cos() + -x.sin())

The function returned by vjp() will compute the partials with respect to each of the primals

>>> x, y = torch.randn([5, 4]), torch.randn([4, 5])
>>> (_, vjpfunc) = torch.func.vjp(torch.matmul, x, y)
>>> cotangents = torch.randn([5, 5])
>>> vjps = vjpfunc(cotangents)
>>> assert len(vjps) == 2
>>> assert torch.allclose(vjps[0], torch.matmul(cotangents, y.transpose(0, 1)))
>>> assert torch.allclose(vjps[1], torch.matmul(x.transpose(0, 1), cotangents))

primals are the positional arguments for f. All kwargs use their default value

>>> x = torch.randn([5])
>>> def f(x, scale=4.):
>>>   return x * scale
>>>
>>> (_, vjpfunc) = torch.func.vjp(f, x)
>>> vjps = vjpfunc(torch.ones_like(x))
>>> assert torch.allclose(vjps[0], torch.full(x.shape, 4.0))

Note

Using PyTorch torch.no_grad together with vjp. Case 1: Using torch.no_grad inside a function:

>>> def f(x):
>>>     with torch.no_grad():
>>>         c = x ** 2
>>>     return x - c

In this case, vjp(f)(x) will respect the inner torch.no_grad.

Case 2: Using vjp inside torch.no_grad context manager:

>>> # xdoctest: +SKIP(failing)
>>> with torch.no_grad():
>>>     vjp(f)(x)

In this case, vjp will respect the inner torch.no_grad, but not the outer one. This is because vjp is a “function transform”: its result should not depend on the result of a context manager outside of f.

effectful.handlers.torch.vmap(func: Callable[[_P], _R], in_dims: int | tuple[Any, ...] = 0, out_dims: int | tuple[int, ...] | None = 0, randomness: str = 'error', *, chunk_size: int | None = None) → Callable[[_P], _R]

vmap is the vectorizing map; vmap(func) returns a new function that maps func over some dimension of the inputs. Semantically, vmap pushes the map into PyTorch operations called by func, effectively vectorizing those operations.

vmap is useful for handling batch dimensions: one can write a function func that runs on examples and then lift it to a function that can take batches of examples with vmap(func). vmap can also be used to compute batched gradients when composed with autograd.

Note

torch.vmap() is aliased to torch.func.vmap() for convenience. Use whichever one you’d like.

Args:

func (function): A Python function that takes one or more arguments.: Must return one or more Tensors.
in_dims (int or nested structure): Specifies which dimension of the: inputs should be mapped over. in_dims should have a structure like the inputs. If the in_dim for a particular input is None, then that indicates there is no map dimension. Default: 0.
out_dims (int or Tuple[int]): Specifies where the mapped dimension: should appear in the outputs. If out_dims is a Tuple, then it should have one element per output. Default: 0.
randomness (str): Specifies whether the randomness in this: vmap should be the same or different across batches. If ‘different’, the randomness for each batch will be different. If ‘same’, the randomness will be the same across batches. If ‘error’, any calls to random functions will error. Default: ‘error’. WARNING: this flag only applies to random PyTorch operations and does not apply to Python’s random module or numpy randomness.
chunk_size (None or int): If None (default), apply a single vmap over inputs.: If not None, then compute the vmap chunk_size samples at a time. Note that chunk_size=1 is equivalent to computing the vmap with a for-loop. If you run into memory issues computing the vmap, please try a non-None chunk_size.

Returns:

Returns a new “batched” function. It takes the same inputs as func, except each input has an extra dimension at the index specified by in_dims. It takes returns the same outputs as func, except each output has an extra dimension at the index specified by out_dims.

One example of using vmap() is to compute batched dot products. PyTorch doesn’t provide a batched torch.dot API; instead of unsuccessfully rummaging through docs, use vmap() to construct a new function.

>>> torch.dot  # [D], [D] -> []
>>> batched_dot = torch.func.vmap(torch.dot)  # [N, D], [N, D] -> [N]
>>> x, y = torch.randn(2, 5), torch.randn(2, 5)
>>> batched_dot(x, y)

vmap() can be helpful in hiding batch dimensions, leading to a simpler model authoring experience.

>>> batch_size, feature_size = 3, 5
>>> weights = torch.randn(feature_size, requires_grad=True)
>>>
>>> def model(feature_vec):
>>> # Very simple linear model with activation
>>>     return feature_vec.dot(weights).relu()
>>>
>>> examples = torch.randn(batch_size, feature_size)
>>> result = torch.vmap(model)(examples)

vmap() can also help vectorize computations that were previously difficult or impossible to batch. One example is higher-order gradient computation. The PyTorch autograd engine computes vjps (vector-Jacobian products). Computing a full Jacobian matrix for some function f: R^N -> R^N usually requires N calls to autograd.grad, one per Jacobian row. Using vmap(), we can vectorize the whole computation, computing the Jacobian in a single call to autograd.grad.

>>> # Setup
>>> N = 5
>>> f = lambda x: x**2
>>> x = torch.randn(N, requires_grad=True)
>>> y = f(x)
>>> I_N = torch.eye(N)
>>>
>>> # Sequential approach
>>> jacobian_rows = [torch.autograd.grad(y, x, v, retain_graph=True)[0]
>>>                  for v in I_N.unbind()]
>>> jacobian = torch.stack(jacobian_rows)
>>>
>>> # vectorized gradient computation
>>> def get_vjp(v):
>>>     return torch.autograd.grad(y, x, v)
>>> jacobian = torch.vmap(get_vjp)(I_N)

vmap() can also be nested, producing an output with multiple batched dimensions

>>> torch.dot  # [D], [D] -> []
>>> batched_dot = torch.vmap(
...     torch.vmap(torch.dot)
... )  # [N1, N0, D], [N1, N0, D] -> [N1, N0]
>>> x, y = torch.randn(2, 3, 5), torch.randn(2, 3, 5)
>>> batched_dot(x, y)  # tensor of size [2, 3]

If the inputs are not batched along the first dimension, in_dims specifies the dimension that each inputs are batched along as

>>> torch.dot  # [N], [N] -> []
>>> batched_dot = torch.vmap(torch.dot, in_dims=1)  # [N, D], [N, D] -> [D]
>>> x, y = torch.randn(2, 5), torch.randn(2, 5)
>>> batched_dot(
...     x, y
... )  # output is [5] instead of [2] if batched along the 0th dimension

If there are multiple inputs each of which is batched along different dimensions, in_dims must be a tuple with the batch dimension for each input as

>>> torch.dot  # [D], [D] -> []
>>> batched_dot = torch.vmap(torch.dot, in_dims=(0, None))  # [N, D], [D] -> [N]
>>> x, y = torch.randn(2, 5), torch.randn(5)
>>> batched_dot(
...     x, y
... )  # second arg doesn't have a batch dim because in_dim[1] was None

If the input is a Python struct, in_dims must be a tuple containing a struct matching the shape of the input:

>>> f = lambda dict: torch.dot(dict["x"], dict["y"])
>>> x, y = torch.randn(2, 5), torch.randn(5)
>>> input = {"x": x, "y": y}
>>> batched_dot = torch.vmap(f, in_dims=({"x": 0, "y": None},))
>>> batched_dot(input)

By default, the output is batched along the first dimension. However, it can be batched along any dimension by using out_dims

>>> f = lambda x: x**2
>>> x = torch.randn(2, 5)
>>> batched_pow = torch.vmap(f, out_dims=1)
>>> batched_pow(x)  # [5, 2]

For any function that uses kwargs, the returned function will not batch the kwargs but will accept kwargs

>>> x = torch.randn([2, 5])
>>> def fn(x, scale=4.):
>>>   return x * scale
>>>
>>> batched_pow = torch.vmap(fn)
>>> assert torch.allclose(batched_pow(x), x * 4)
>>> batched_pow(x, scale=x)  # scale is not batched, output has shape [2, 2, 5]

Note

vmap does not provide general autobatching or handle variable-length sequences out of the box.

Return type:: Callable[[ParamSpec(_P, bound= None)], TypeVar(_R)]

effectful.handlers.torch.bind_dims(value: Scoped(ordinal=frozenset({B, A}))], *names: Scoped(ordinal=frozenset({B}))]) → Scoped(ordinal=frozenset({A}))][source]

Convert named dimensions to positional dimensions.

Parameters:

t – A tensor.
args – Named dimensions to convert to positional dimensions. These positional dimensions will appear at the beginning of the shape.

Return type:

TypeVar(HasDims, bound= Any | Tensor | Distribution)

Returns:

A tensor with the named dimensions in args converted to positional dimensions.

Example usage:

>>> a, b = defop(torch.Tensor, name='a'), defop(torch.Tensor, name='b')
>>> t = torch.ones(2, 3)
>>> bind_dims(t[a(), b()], b, a).shape
torch.Size([3, 2])

effectful.handlers.torch.sizesof(value) → Mapping[Operation[(), Tensor], int][source]

Return the sizes of named dimensions in a tensor expression.

Sizes are inferred from the tensor shape.

Parameters:: value – A tensor expression.
Return type:: Mapping[Operation[(), Tensor], int]
Returns:: A mapping from named dimensions to their sizes.

Example usage:

>>> a, b = defop(torch.Tensor, name='a'), defop(torch.Tensor, name='b')
>>> sizes = sizesof(torch.ones(2, 3)[a(), b()])
>>> assert sizes[a] == 2 and sizes[b] == 3

effectful.handlers.torch.torch_getitem(*args, **kwargs) → Tensor[source]

Operation for indexing a tensor.

Note

This operation is not intended to be called directly. Instead, it is exposed so that it can be handled.

Return type:: Tensor

effectful.handlers.torch.unbind_dims(value: Scoped(ordinal=frozenset({B, A}))], *names: Scoped(ordinal=frozenset({B}))]) → Scoped(ordinal=frozenset({B, A}))][source]

Return type:: TypeVar(HasDims, bound= Any | Tensor | Distribution)

Indexed

class effectful.handlers.indexed.IndexSet(**mapping: int | Iterable[int])[source]

IndexSet s represent the support of an indexed value, for which free variables correspond to single interventions and indices to worlds where that intervention either did or did not happen.

IndexSet can be understood conceptually as generalizing torch.Size from multidimensional arrays to arbitrary values, from positional to named dimensions, and from bounded integer interval supports to finite sets of positive integers.

IndexSet`s are implemented as :class:`dict`s with :class:`str`s as keys corresponding to names of free index variables and :class:`set s of positive int s as values corresponding to the values of the index variables where the indexed value is defined.

For example, the following IndexSet represents the sets of indices of the free variables x and y for which a value is defined:

>>> IndexSet(x={0, 1}, y={2, 3})
IndexSet({'x': {0, 1}, 'y': {2, 3}})

IndexSet ‘s constructor will automatically drop empty entries and attempt to convert input values to set s:

>>> IndexSet(x=[0, 0, 1], y=set(), z=2)
IndexSet({'x': {0, 1}, 'z': {2}})

IndexSet s are also hashable and can be used as keys in dict s:

>>> indexset = IndexSet(x={0, 1}, y={2, 3})
>>> indexset in {indexset: 1}
True

effectful.handlers.indexed.cond(fst: Tensor, snd: Tensor, case_: Tensor) → Tensor[source]

Selection operation that is the sum-type analogue of scatter() in the sense that where scatter() propagates both of its arguments, cond() propagates only one, depending on the value of a boolean case .

For a given fst , snd , and case , cond() returns snd if the case is true, and fst otherwise, analogous to a Python conditional expression snd if case else fst . Unlike a Python conditional expression, however, the case may be a tensor, and both branches are evaluated, as with torch.where()

>>> from effectful.ops.syntax import defop
>>> from effectful.handlers.torch import bind_dims

>>> b = defop(torch.Tensor, name="b")
>>> fst, snd = torch.randn(2, 3)[b()], torch.randn(2, 3)[b()]
>>> case = (fst < snd).all(-1)
>>> x = cond(fst, snd, case)
>>> assert (bind_dims(x, b) == bind_dims(torch.where(case[..., None], snd, fst), b)).all()

Note

cond() can be extended to new value types by registering an implementation for the type using functools.singledispatch() .

Parameters:

fst (Tensor) – The value to return if case is False .
snd (Tensor) – The value to return if case is True .
case – A boolean value or tensor. If a tensor, should have event shape () .

Return type:

Tensor

effectful.handlers.indexed.cond_n(values: dict[IndexSet, Tensor], case: Tensor) → Tensor[source]

Return type:: Tensor

effectful.handlers.indexed.gather(value: Tensor, indexset: IndexSet) → Tensor[source]

Selects entries from an indexed value at the indices in a IndexSet . gather() is useful in conjunction with MultiWorldCounterfactual for selecting components of a value corresponding to specific counterfactual worlds.

For example, in a model with an outcome variable Y and a treatment variable T that has been intervened on, we can use gather() to define quantities like treatment effects that require comparison of different potential outcomes:

>>> def example():
...     with MultiWorldCounterfactual():
...         X = pyro.sample("X", get_X_dist())
...         T = pyro.sample("T", get_T_dist(X))
...         T = intervene(T, t, name="T_ax")  # adds an index variable "T_ax"
...         Y = pyro.sample("Y", get_Y_dist(X, T))
...         Y_factual = gather(Y, IndexSet(T_ax=0))         # no intervention
...         Y_counterfactual = gather(Y, IndexSet(T_ax=1))  # intervention
...         treatment_effect = Y_counterfactual - Y_factual
>>> example()

Like torch.gather() and substitution in term rewriting, gather() is defined extensionally, meaning that values are treated as constant functions of variables not in their support.

gather() will accordingly ignore variables in indexset that are not in the support of value computed by indices_of() .

Note

gather() can be extended to new value types by registering an implementation for the type using functools.singledispatch() .

Note

Fully general versions of indices_of() , gather() and scatter() would require a dependent broadcasting semantics for indexed values, as is the case in sparse or masked array libraries like scipy.sparse or xarray or in relational databases.

However, this is beyond the scope of this library as it currently exists. Instead, gather() currently binds free variables in indexset when their indices there are a strict subset of the corresponding indices in value , so that they no longer appear as free in the result.

For example, in the above snippet, applying gather() to to select only the values of Y from worlds where no intervention on T happened would result in a value that no longer contains free variable "T":

>>> indices_of(Y) == IndexSet(T_ax={0, 1})
True
>>> Y0 = gather(Y, IndexSet(T_ax={0}))
>>> indices_of(Y0) == IndexSet() != IndexSet(T_ax={0})
True

The practical implications of this imprecision are limited since we rarely need to gather() along a variable twice.

Parameters:

value (Tensor) – The value to gather.
indexset (IndexSet) – The IndexSet of entries to select from value.

Return type:

Tensor

Returns:

A new value containing entries of value from indexset.

effectful.handlers.indexed.indices_of(value: Any) → IndexSet[source]

Get a IndexSet of indices on which an indexed value is supported. indices_of() is useful in conjunction with MultiWorldCounterfactual for identifying the worlds where an intervention happened upstream of a value.

For example, in a model with an outcome variable Y and a treatment variable T that has been intervened on, T and Y are both indexed by "T":

>>> def example():
...     with MultiWorldCounterfactual():
...         X = pyro.sample("X", get_X_dist())
...         T = pyro.sample("T", get_T_dist(X))
...         T = intervene(T, t, name="T_ax")  # adds an index variable "T_ax"
...         Y = pyro.sample("Y", get_Y_dist(X, T))
...         assert indices_of(X) == IndexSet({})
...         assert indices_of(T) == IndexSet({T_ax: {0, 1}})
...         assert indices_of(Y) == IndexSet({T_ax: {0, 1}})
>>> example()

Just as multidimensional arrays can be expanded to shapes with new dimensions over which they are constant, indices_of() is defined extensionally, meaning that values are treated as constant functions of free variables not in their support.

Note

indices_of() can be extended to new value types by registering an implementation for the type using functools.singledispatch() .

Note

Fully general versions of indices_of() , gather() and scatter() would require a dependent broadcasting semantics for indexed values, as is the case in sparse or masked array libraries like torch.sparse or relational databases.

However, this is beyond the scope of this library as it currently exists. Instead, gather() currently binds free variables in its input indices when their indices there are a strict subset of the corresponding indices in value , so that they no longer appear as free in the result.

For example, in the above snippet, applying gather() to to select only the values of Y from worlds where no intervention on T happened would result in a value that no longer contains free variable "T":

>>> indices_of(Y) == IndexSet(T_ax={0, 1})
True
>>> Y0 = gather(Y, IndexSet(T_ax={0}))
>>> indices_of(Y0) == IndexSet() != IndexSet(T_ax={0})
True

The practical implications of this imprecision are limited since we rarely need to gather() along a variable twice.

Parameters:

value (Any) – A value.
kwargs – Additional keyword arguments used by specific implementations.

Return type:

IndexSet

Returns:

A IndexSet containing the indices on which the value is supported.

effectful.handlers.indexed.name_to_sym(name: str) → Operation[(), Tensor][source]

Return type:: Operation[(), Tensor]

effectful.handlers.indexed.stack(values: tuple[Tensor, ...] | list[Tensor], name: str) → Tensor[source]

Stack a sequence of indexed values, creating a new dimension. The new dimension is indexed by dim. The indexed values in the stack must have identical shapes.

Return type:: Tensor

effectful.handlers.indexed.union(*indexsets: IndexSet) → IndexSet[source]

Compute the union of multiple IndexSet s as the union of their keys and of value sets at shared keys.

If IndexSet may be viewed as a generalization of torch.Size, then union() is a generalization of torch.broadcast_shapes() for the more abstract IndexSet data structure.

Example:

>>> s = union(IndexSet(a={0, 1}, b={1}), IndexSet(a={1, 2}))
>>> s["a"]
{0, 1, 2}
>>> s["b"]
{1}

Note

union() satisfies several algebraic equations for arbitrary inputs. In particular, it is associative, commutative, idempotent and absorbing:

union(a, union(b, c)) == union(union(a, b), c)
union(a, b) == union(b, a)
union(a, a) == a
union(a, union(a, b)) == union(a, b)

Return type:: IndexSet

Internals

Runtime

type effectful.internals.runtime.CacheEntry = AutoIdKeyDictionary[Interpretation, Any]

type effectful.internals.runtime.EvalCache = AutoIdKeyDictionary[Any, CacheEntry]

effectful.internals.runtime.cache(store: EvalCache | None = None)[source]

Memoize evaluation under any interpretation for the duration of this block.

Installs store, or a fresh cache if none is given, and yields it so that a later block can reuse it:

with cache() as store:
    ...
with cache(store):
    ...

effectful.ops.semantics.evaluate() installs one for the duration of a call when none is active, so a lone call is memoized internally whether or not a scope is open. Holding a scope is what shares that work between calls.

effectful.internals.runtime.cache_get(store: EvalCache, expr: Any, intp: Interpretation, default: Any = None) → Any[source]

Look expr up under intp, returning default if it is not cached.

Return type:: Any

effectful.internals.runtime.cache_put(store: EvalCache, expr: Any, intp: Interpretation, value: Any) → None[source]

Record that expr evaluates to value under intp.

Return type:: None

effectful.internals.runtime.copy_cache_entries(src, dst) → None[source]

Copy everything cached for src onto dst.

effectful.ops.syntax._build_term() computes a node’s type analysis on a throwaway term and then needs it attributed to the term it actually returns.

Return type:: None

effectful.internals.runtime.get_interpretation()

Return a value for the context variable for the current context.

If there is no value for the variable in the current context, the method will:

return the value of the default argument of the method, if provided; or

return the default value for the context variable, if it was created with one; or

raise a LookupError.

effectful.internals.runtime.interpreter(intp: Interpretation)[source]

Unification

Type unification and inference utilities for Python’s generic type system.

This module implements a unification algorithm for type inference over a subset of Python’s generic types. Unification is a fundamental operation in type systems that finds substitutions for type variables to make two types equivalent.

The module provides four main operations:

unify(typ, subtyp, subs={}): The core unification algorithm that attempts to find a substitution mapping for type variables that makes a pattern type equal to a concrete type. It handles TypeVars, generic types (List[T], Dict[K,V]), unions, callables, and function signatures with inspect.Signature/BoundArguments.
substitute(typ, subs): Applies a substitution mapping to a type expression, replacing all TypeVars with their mapped concrete types. This is used to instantiate generic types after unification.
freetypevars(typ): Extracts all free (unbound) type variables from a type expression. Useful for analyzing generic types and ensuring all TypeVars are properly bound.
nested_type(value): Infers the type of a runtime value, handling nested collections by recursively determining element types. For example, [1, 2, 3] becomes list[int], and {“key”: [1, 2]} becomes dict[str, list[int]].

The unification algorithm uses a single-dispatch pattern to handle different type combinations: - TypeVar unification binds variables to concrete types - Generic type unification matches origins and recursively unifies type arguments - Structural unification handles sequences and mappings by element - Union types attempt unification with any matching branch - Function signatures unify parameter types with bound arguments

Example usage:

>>> from effectful.internals.unification import unify, substitute, freetypevars
>>> import typing
>>> T = typing.TypeVar('T')
>>> K = typing.TypeVar('K')
>>> V = typing.TypeVar('V')

>>> # Find substitution that makes list[T] equal to list[int]
>>> subs = unify(list[T], list[int])
>>> subs
{~T: <class 'int'>}

>>> # Apply substitution to instantiate a generic type
>>> substitute(dict[K, list[V]], {K: str, V: int})
dict[str, list[int]]

>>> # Find all type variables in a type expression
>>> freetypevars(dict[str, list[V]])
{~V}

This module is primarily used internally by effectful for type inference in its effect system, allowing it to track and propagate type information through effect handlers and operations.

class effectful.internals.unification.Box(value: T) → None[source]

Boxed types. Prevents confusion between types computed by __type_rule__ and values.

value: T

class effectful.internals.unification.TypeEvaluator[source]

Abstract base class for evaluating type expressions.

This class defines the interface for evaluating type expressions, which may involve resolving type variables, computing canonical forms of types, or performing other transformations. Subclasses should implement the evaluate method to provide specific evaluation logic.

The TypeEvaluator can be used in contexts where type expressions need to be processed or normalized before unification or other type operations.

evaluate(typ: GenericAlias | _GenericAlias)

evaluate(typ: Union)

evaluate(typ: _AnnotatedAlias)

evaluate(typ: _LiteralGenericAlias)

evaluate(typ: ParamSpecArgs | ParamSpecKwargs)

evaluate(typ: _SpecialGenericAlias)

evaluate(typ: _ConcatenateGenericAlias)

evaluate(typ: list | tuple)

evaluate(typ: NewType)

evaluate(typ: TypeAliasType)

evaluate(typ: ForwardRef)

Normalize generic types

Return type:: TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union | Sequence[TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union]

effectful.internals.unification.canonicalize(typ: type | ABCMeta)

effectful.internals.unification.canonicalize(typ: EllipsisType | None)

effectful.internals.unification.canonicalize(typ: TypeVar)

effectful.internals.unification.canonicalize(typ: ParamSpec)

effectful.internals.unification.canonicalize(typ: TypeVarTuple)

effectful.internals.unification.canonicalize(typ: Union)

effectful.internals.unification.canonicalize(typ: GenericAlias | _GenericAlias)

effectful.internals.unification.canonicalize(typ: list | tuple)

effectful.internals.unification.canonicalize(typ: _InterpretationMeta)

effectful.internals.unification.canonicalize(typ: _AnnotatedAlias)

effectful.internals.unification.canonicalize(typ: _SpecialGenericAlias)

effectful.internals.unification.canonicalize(typ: _LiteralGenericAlias)

effectful.internals.unification.canonicalize(typ: NewType)

effectful.internals.unification.canonicalize(typ: TypeAliasType)

effectful.internals.unification.canonicalize(typ: _ConcatenateGenericAlias)

effectful.internals.unification.canonicalize(typ: _AnyMeta)

effectful.internals.unification.canonicalize(typ: ParamSpecArgs | ParamSpecKwargs)

effectful.internals.unification.canonicalize(typ: _SpecialForm)

effectful.internals.unification.canonicalize(typ: _ProtocolMeta)

effectful.internals.unification.canonicalize(typ: _UnpackGenericAlias)

effectful.internals.unification.canonicalize(typ: ForwardRef)

Normalize generic types

Return type:: TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union | Sequence[TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union]

effectful.internals.unification.freetypevars(typ) → Set[TypeVar | TypeVarTuple | ParamSpec][source]

Return a set of free type variables in the given type expression.

This function recursively traverses a type expression to find all TypeVar instances that appear within it. It handles both simple types and generic type aliases with nested type arguments. TypeVars are considered “free” when they are not bound to a specific concrete type.

Return type:: Set[TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>)]

Args:

typ: The type expression to analyze. Can be a plain type (e.g., int),: a TypeVar, or a generic type alias (e.g., List[T], Dict[K, V]).

Returns:

A set containing all TypeVar instances found in the type expression. Returns an empty set if no TypeVars are present.

Examples:

>>> T = typing.TypeVar('T')
>>> K = typing.TypeVar('K')
>>> V = typing.TypeVar('V')

>>> # TypeVar returns itself
>>> freetypevars(T)
{~T}

>>> # Generic type with one TypeVar
>>> freetypevars(list[T])
{~T}

>>> # Generic type with multiple TypeVars
>>> freetypevars(dict[K, V]) == {K, V}
True

>>> # Nested generic types
>>> freetypevars(list[dict[K, V]]) == {K, V}
True

>>> # Concrete types have no free TypeVars
>>> freetypevars(int)
set()

>>> # Generic types with concrete arguments have no free TypeVars
>>> freetypevars(list[int])
set()

>>> # Mixed concrete and TypeVar arguments
>>> freetypevars(dict[str, T])
{~T}

Substitute type variables in a type expression with concrete types.

This function recursively traverses a type expression and replaces any TypeVar instances found with their corresponding concrete types from the substitution mapping. If a TypeVar is not present in the substitution mapping, it remains unchanged. The function handles nested generic types by recursively substituting in their type arguments.

Return type:: TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union | Sequence[TypeVar(TypeVar, bound= <attribute ‘__bound__’ of ‘typing.TypeVar’ objects>, covariant=<member ‘__covariant__’ of ‘typing.TypeVar’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.TypeVar’ objects>) | TypeVarTuple | ParamSpec(ParamSpec, bound= <member ‘__bound__’ of ‘typing.ParamSpec’ objects>, covariant=<member ‘__covariant__’ of ‘typing.ParamSpec’ objects>, contravariant=<member ‘__contravariant__’ of ‘typing.ParamSpec’ objects>) | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union]

Args:

typ: The type expression to perform substitution on. Can be a plain type,: a TypeVar, or a generic type alias (e.g., List[T], Dict[K, V]).
subs: A mapping from TypeVar instances to concrete types that should: replace them.

Returns:

A new type expression with all mapped TypeVars replaced by their corresponding concrete types.

Examples:

>>> T = typing.TypeVar('T')
>>> K = typing.TypeVar('K')
>>> V = typing.TypeVar('V')

>>> # Simple TypeVar substitution
>>> substitute(T, {T: int})
<class 'int'>

>>> # Generic type substitution
>>> substitute(list[T], {T: str})
list[str]

>>> # Nested generic substitution
>>> substitute(dict[K, list[V]], {K: str, V: int})
dict[str, list[int]]

>>> # TypeVar not in mapping remains unchanged
>>> substitute(T, {K: int})
~T

>>> # Non-generic types pass through unchanged
>>> substitute(int, {T: str})
<class 'int'>

effectful.internals.unification.unify(typ, subtyp, subs: Mapping[TypeVar | TypeVarTuple | ParamSpec, TypeVar | TypeVarTuple | ParamSpec | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union | Sequence[TypeVar | TypeVarTuple | ParamSpec | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union]] = {}) → Mapping[TypeVar | TypeVarTuple | ParamSpec, TypeVar | TypeVarTuple | ParamSpec | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union | Sequence[TypeVar | TypeVarTuple | ParamSpec | type | ABCMeta | EllipsisType | None | _AnyMeta | GenericAlias | _GenericAlias | Union]][source]

Overloads:

typ (inspect.Signature), subtyp (inspect.BoundArguments), subs (Substitutions) → Substitutions
typ (TypeExpressions), subtyp (TypeExpressions), subs (Substitutions) → Substitutions

Unify a pattern type with a concrete type, returning a substitution map.

This function attempts to find a substitution of type variables that makes the pattern type (typ) equal to the concrete type (subtyp). It updates and returns the substitution mapping, or raises TypeError if unification is not possible.

The function handles: - TypeVar unification (binding type variables to concrete types) - Generic type unification (matching origins and recursively unifying args) - Structural unification of sequences and mappings - Exact type matching for non-generic types

Args:

typ: The pattern type that may contain TypeVars to be unified subtyp: The concrete type to unify with the pattern subs: Existing substitution mappings to be extended (not modified)

Returns:

A new substitution mapping that includes all previous substitutions plus any new TypeVar bindings discovered during unification.

Raises:

TypeError: If unification is not possible (incompatible types or: conflicting TypeVar bindings)

Examples:

>>> import typing
>>> T = typing.TypeVar('T')
>>> K = typing.TypeVar('K')
>>> V = typing.TypeVar('V')

>>> # Simple TypeVar unification
>>> unify(T, int, {})
{~T: <class 'int'>}

>>> # Generic type unification
>>> unify(list[T], list[int], {})
{~T: <class 'int'>}

>>> # Exact type matching
>>> unify(int, int, {})
{}

>>> # Failed unification - incompatible types
>>> unify(list[T], dict[str, int], {})
Traceback (most recent call last):
    ...
TypeError: Cannot unify ...

>>> # Failed unification - conflicting TypeVar binding
>>> unify(T, str, {T: int})
Traceback (most recent call last):
    ...
TypeError: Cannot unify ...