Queries and Prompting Policies

Queries

Query

Bases: AbstractQuery[T]

Base class for queries.

This class adds standard convenience features on top of AbstractQuery, using reflection to allow queries to be defined concisely. Here is a simple example of a query type definition:

@dataclass class MakeSum(Query[list[int]]):
    '''Given a list of allowed numbers and a target number, you
    must find a sub-list whose elements sum up to the target.
    Just answer with a list of numbers as a JSON object and
    nothing else.'''

    allowed: list[int] target: int

In general, a query type is a dataclass that inherits Query[T], where T is the query's answer type. In the example above, no parser is specified and so oracles are requested to provide structured answers as JSON objects, which are automatically parsed into the answer type (list[int]) using pydantic. Assuming that no Jinja prompt file is provided, the docstring is used as a system prompt and instance prompts are generated by simply serializing MakeSum instances into YAML.

All attributes of a query must be serializable by pydantic. They can be builtin types (int, list, dict...), custom dataclasses...

Customizing Prompts

System and instance prompts can be specified via Jinja templates. The templates manager (TemplatesManager) looks for templates named "..jinja". Templates can also be provided by defining the __system_prompt__ and/or __instance_prompt__ class attributes. If none of these are provided, the query's docstring is used as a system prompt and DEFAULT_INSTANCE_PROMPT is used as an instance prompt template. All attributes from QueryTemplateArgs are made available to templates, with possibly extra ones.

Answer Modes and Configurations

A query can define several answer modes (AnswerMode), each of which can be associated with a different parser and set of settings. By default, the only answer mode is None. More answer modes can be defined by setting class variable __modes__.

The parser_for method maps modes to parser specifications. Its default implementation first checks whether the parser method is overriden, in which case it is used. Otherwise, the __parser__ attribute is checked. If none of these conditions hold, structured is used as a default parser.

Allowing Multi-Message Exchanges and Tool Calls

A common pattern for interacting with LLMs is to have multi-message exchanges where the full conversation history is resent repeatedly. LLMs are also often allowed to request tool call. This interaction pattern is implemented in the interact standard strategy. It is enabled by several features on the Query side.

Answer Prefixes

If a query type has a prefix attribute with type AnswerPrefix, this attribute can be used to provide a chat history, to be added to the query's prompt.

The Response Type

If the query answer type is Response, the query does not only return a parsed answer, but also the LLM raw answer (which can be appended to a chat history), and possibly a sequence of tool calls.

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

class Query[T](dp.AbstractQuery[T]):
    """
    Base class for queries.

    This class adds standard convenience features on top of
    `AbstractQuery`, using reflection to allow queries to be defined
    concisely. Here is a simple example of a query type definition:

    ```python
    @dataclass class MakeSum(Query[list[int]]):
        '''Given a list of allowed numbers and a target number, you
        must find a sub-list whose elements sum up to the target.
        Just answer with a list of numbers as a JSON object and
        nothing else.'''

        allowed: list[int] target: int
    ```

    In general, a query type is a dataclass that inherits `Query[T]`,
    where `T` is the query's answer type. In the example above, no
    parser is specified and so oracles are requested to provide
    structured answers as JSON objects, which are automatically parsed
    into the answer type (`list[int]`) using pydantic. Assuming that no
    Jinja prompt file is provided, the docstring is used as a system
    prompt and instance prompts are generated by simply serializing
    `MakeSum` instances into YAML.

    All attributes of a query must be serializable by pydantic. They can
    be builtin types (int, list, dict...), custom dataclasses...

    ## Customizing Prompts

    System and instance prompts can be specified via Jinja templates.
    The templates manager (`TemplatesManager`) looks for templates named
    "<QueryName>.<instance|system>.jinja". Templates can also be
    provided by defining the `__system_prompt__` and/or
    `__instance_prompt__` class attributes. If none of these are
    provided, the query's docstring is used as a system prompt and
    `DEFAULT_INSTANCE_PROMPT` is used as an instance prompt template.
    All attributes from `QueryTemplateArgs` are made available to
    templates, with possibly extra ones.

    ## Answer Modes and Configurations

    A query can define several answer modes (`AnswerMode`), each of
    which can be associated with a different parser and set of settings.
    By default, the only answer mode is `None`. More answer modes can be
    defined by setting class variable `__modes__`.

    The `parser_for` method maps modes to parser specifications. Its
    default implementation first checks whether the `parser` method is
    overriden, in which case it is used. Otherwise, the `__parser__`
    attribute is checked. If none of these conditions hold, `structured`
    is used as a default parser.

    ## Allowing Multi-Message Exchanges and Tool Calls

    A common pattern for interacting with LLMs is to have multi-message
    exchanges where the full conversation history is resent repeatedly.
    LLMs are also often allowed to request tool call. This interaction
    pattern is implemented in the `interact` standard strategy. It is
    enabled by several features on the `Query` side.

    ### Answer Prefixes

    If a query type has a prefix attribute with type `AnswerPrefix`,
    this attribute can be used to provide a chat history, to be added to
    the query's prompt.

    ### The `Response` Type

    If the query answer type is `Response`, the query does not only
    return a parsed answer, but also the LLM raw answer (which can be
    appended to a chat history), and possibly a sequence of tool calls.
    """

    __modes__: ClassVar[Sequence[dp.AnswerMode] | None] = None
    __parser__: ClassVar[Parser[Any] | GenericParser | ParserDict | None] = (
        None
    )

    ### Parsing Answers

    def parser(self) -> Parser[T] | GenericParser:
        """
        Method to override to provide a parser specification common to
        all modes. Alternatively, the `__parser__` class attribute can
        be set. The first method allows more flexibility since parser
        specifications can then depend on query attributes.
        """
        assert False, (
            "Please provide `__parser__`, `parser` or "
            + f"`parser_for` for query type {type(self)}"
        )

    def parser_for(self, mode: dp.AnswerMode) -> Parser[T] | GenericParser:
        """
        Obtain a parser speficiation for a given answer mode.

        This method can be overriden. By default, it does the following:

        1. If the `parser` method is overriden, it uses it.
        2. If `__parser__` is set as a parser, it is used.
        2. If `__parser__` is set as a dictionary, the mode is used as a
           key to obtain a parser.
        3. Otherwise, `structured` is used as a default parser.
        """
        if dpi.is_method_overridden(Query, type(self), "parser"):
            assert self.__parser__ is None, (
                f"Both `__parser__` and `parser` are provided for {type(self)}."
            )
            return self.parser()
        elif self.__parser__ is None:
            return structured  # default parser
        else:
            assert not dpi.is_method_overridden(
                Query, type(self), "parser_for"
            ), (
                "Both `__parser__` and `parser_for` are "
                + f"provided for {type(self)}."
            )
            parser_attr = self.__parser__
            if isinstance(parser_attr, dict):
                parser = parser_attr[mode]
            else:
                parser = parser_attr
            assert isinstance(parser, (Parser, GenericParser)), (
                "Expected parser type, got: " + f"{type(parser)}."
            )
            return cast(Any, parser)

    def _instantiated_parser_for(self, mode: dp.AnswerMode) -> Parser[T]:
        parser = self.parser_for(mode)
        if isinstance(parser, GenericParser):
            return parser.for_type(self._answer_type())
        else:
            return parser

    @override
    def parse_answer(self, answer: dp.Answer) -> T | dp.ParseError:
        assert answer.mode in self.query_modes(), (
            f"Unknown mode: {answer.mode}"
        )
        try:
            parser = self._instantiated_parser_for(answer.mode)
            return parser.parse(answer)
        except dp.ParseError as e:
            return e

    @override
    def query_settings(self, mode: dp.AnswerMode) -> dp.QuerySettings:
        parser = self._instantiated_parser_for(mode)
        return parser.settings

    ### Query Prefixes

    @classmethod
    def _has_special_prefix_attr(cls):
        annots = typing.get_type_hints(cls)
        return "prefix" in annots and annots["prefix"] is ct.AnswerPrefix

    @override
    def query_prefix(self) -> ct.AnswerPrefix | None:
        """
        Return the value of the `prefix` attribute if it has type
        annotation `AnswerPrefix` or return `None`.
        """
        if self._has_special_prefix_attr():
            return getattr(self, "prefix")
        return None

    ### Producing Prompts

    @override
    def generate_prompt(
        self,
        *,
        kind: Literal["system", "instance"] | str,
        mode: dp.AnswerMode,
        params: dict[str, object],
        extra_args: dict[str, object] | None = None,
        env: dp.AbstractTemplatesManager | None = None,
    ) -> str:
        assert env is not None, _no_prompt_manager_error()
        args_min: QueryTemplateArgs = {
            "query": self,
            "mode": mode,
            "available_modes": self.query_modes(),
            "params": params,
            "format": self._instantiated_parser_for(mode).formatting,
        }
        args: dict[str, object] = {**args_min}
        if extra_args:
            args.update(extra_args)
        if (glob := self.globals()) is not None:
            args["globals"] = glob
        return env.prompt(
            query_name=self.query_name(),
            prompt_kind=kind,
            template_args=args,
            default_template=self._default_prompt(kind),
        )

    @classmethod
    def _default_prompt(
        cls, kind: Literal["system", "instance"] | str
    ) -> str | None:
        attr_name = f"__{kind}_prompt__"
        if hasattr(cls, attr_name):
            res = getattr(cls, attr_name)
            assert isinstance(res, str)
            return textwrap.dedent(res).strip()
        if kind == "instance":
            if cls._has_special_prefix_attr():
                return DEFAULT_INSTANCE_PROMPT_WITH_PREFIX
            else:
                return DEFAULT_INSTANCE_PROMPT
        if kind == "system" and (doc := inspect.getdoc(cls)) is not None:
            return doc
        if kind == "feedback":
            return DEFAULT_FEEDBACK_PROMPT
        return None

    def globals(self) -> dict[str, object] | None:
        """
        Return global objects accessible in prompts via the `globals`
        attribute.
        """
        return None

    ### Other Simple Overrides

    @override
    def serialize_args(self) -> dict[str, object]:
        return cast(dict[str, object], ty.pydantic_dump(type(self), self))

    @classmethod
    def _answer_type(cls) -> TypeAnnot[T]:
        return dpi.first_parameter_of_base_class(cls)

    @override
    def answer_type(self) -> TypeAnnot[T]:
        return self._answer_type()

    @override
    def finite_answer_set(self) -> Sequence[dp.Answer] | None:
        # We handle the special case where the return type is a literal
        # type that is a subtype of str.
        ans = self.answer_type()
        if (res := _match_string_literal_type(ans)) is not None:
            return [dp.Answer(None, v) for v in res]
        return None

    @override
    def query_modes(self) -> Sequence[dp.AnswerMode]:
        if self.__modes__ is not None:
            return self.__modes__
        return [None]

    ### Generating Opaque Spaces

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

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

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

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

        The optional `inner_policy_type` argument is ignored at runtime
        and can be used to help type checkers infer the type of the
        ambient inner policy.
        """
        if isinstance(get_policy, EllipsisType):
            return OpaqueSpace[P, T].from_query(
                self, cast(Any, pol.dict_subpolicy)
            )
        return OpaqueSpace[P, T].from_query(self, lambda p, _: get_policy(p))

    def run_toplevel(
        self,
        env: PolicyEnv,
        policy: pol.PromptingPolicy,
    ) -> Stream[T]:
        """
        Obtain a search stream of query answers, given a prompting
        policy.
        """
        attached = dp.spawn_standalone_query(self)
        return policy(attached, env)

parser

parser() -> Parser[T] | GenericParser

Method to override to provide a parser specification common to all modes. Alternatively, the __parser__ class attribute can be set. The first method allows more flexibility since parser specifications can then depend on query attributes.

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

def parser(self) -> Parser[T] | GenericParser:
    """
    Method to override to provide a parser specification common to
    all modes. Alternatively, the `__parser__` class attribute can
    be set. The first method allows more flexibility since parser
    specifications can then depend on query attributes.
    """
    assert False, (
        "Please provide `__parser__`, `parser` or "
        + f"`parser_for` for query type {type(self)}"
    )

parser_for

parser_for(mode: AnswerMode) -> Parser[T] | GenericParser

Obtain a parser speficiation for a given answer mode.

This method can be overriden. By default, it does the following:

1. If the parser method is overriden, it uses it.
2. If __parser__ is set as a parser, it is used.
3. If __parser__ is set as a dictionary, the mode is used as a key to obtain a parser.
4. Otherwise, structured is used as a default parser.

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

def parser_for(self, mode: dp.AnswerMode) -> Parser[T] | GenericParser:
    """
    Obtain a parser speficiation for a given answer mode.

    This method can be overriden. By default, it does the following:

    1. If the `parser` method is overriden, it uses it.
    2. If `__parser__` is set as a parser, it is used.
    2. If `__parser__` is set as a dictionary, the mode is used as a
       key to obtain a parser.
    3. Otherwise, `structured` is used as a default parser.
    """
    if dpi.is_method_overridden(Query, type(self), "parser"):
        assert self.__parser__ is None, (
            f"Both `__parser__` and `parser` are provided for {type(self)}."
        )
        return self.parser()
    elif self.__parser__ is None:
        return structured  # default parser
    else:
        assert not dpi.is_method_overridden(
            Query, type(self), "parser_for"
        ), (
            "Both `__parser__` and `parser_for` are "
            + f"provided for {type(self)}."
        )
        parser_attr = self.__parser__
        if isinstance(parser_attr, dict):
            parser = parser_attr[mode]
        else:
            parser = parser_attr
        assert isinstance(parser, (Parser, GenericParser)), (
            "Expected parser type, got: " + f"{type(parser)}."
        )
        return cast(Any, parser)

query_prefix

query_prefix() -> AnswerPrefix | None

Return the value of the prefix attribute if it has type annotation AnswerPrefix or return None.

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

@override
def query_prefix(self) -> ct.AnswerPrefix | None:
    """
    Return the value of the `prefix` attribute if it has type
    annotation `AnswerPrefix` or return `None`.
    """
    if self._has_special_prefix_attr():
        return getattr(self, "prefix")
    return None

globals

globals() -> dict[str, object] | None

Return global objects accessible in prompts via the globals attribute.

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

def globals(self) -> dict[str, object] | None:
    """
    Return global objects accessible in prompts via the `globals`
    attribute.
    """
    return None

using

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

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

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

Turn a query into an opaque space by providing a mapping from the ambient inner policy to a prompting policy.

Attributes:

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

The optional inner_policy_type argument is ignored at runtime and can be used to help type checkers infer the type of the ambient inner policy.

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

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

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

    The optional `inner_policy_type` argument is ignored at runtime
    and can be used to help type checkers infer the type of the
    ambient inner policy.
    """
    if isinstance(get_policy, EllipsisType):
        return OpaqueSpace[P, T].from_query(
            self, cast(Any, pol.dict_subpolicy)
        )
    return OpaqueSpace[P, T].from_query(self, lambda p, _: get_policy(p))

run_toplevel

run_toplevel(env: PolicyEnv, policy: PromptingPolicy) -> Stream[T]

Obtain a search stream of query answers, given a prompting policy.

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

def run_toplevel(
    self,
    env: PolicyEnv,
    policy: pol.PromptingPolicy,
) -> Stream[T]:
    """
    Obtain a search stream of query answers, given a prompting
    policy.
    """
    attached = dp.spawn_standalone_query(self)
    return policy(attached, env)

Parser dataclass

A parser specification.

In addition to a mapping from answers to answer type A, a parser also specifies query settings to be passed to oracles, along with special formatting instructions to be rendered into the prompt. Indeed, these components are typically tied and so specifying them together in a single place is clearer.

Attributes:

Name	Type	Description
settings	QuerySettings	The query settings associated with the parser.
formatting	FormattingMetadata	Formatting metadata.
parse	Callable[[Answer], A]	The parsing function, which is allowed to raise the ParseError exception.

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

@dataclass(frozen=True)
class Parser[A]:
    """
    A parser specification.

    In addition to a mapping from answers to answer type `A`, a parser
    also specifies query settings to be passed to oracles, along with
    special formatting instructions to be rendered into the prompt.
    Indeed, these components are typically tied and so specifying them
    together in a single place is clearer.

    Attributes:
        settings: The query settings associated with the parser.
        formatting: Formatting metadata.
        parse: The parsing function, which is allowed to raise
            the `ParseError` exception.
    """

    settings: dp.QuerySettings
    formatting: FormattingMetadata
    parse: Callable[[dp.Answer], A]

    def update_formatting(
        self, f: Callable[[FormattingMetadata], FormattingMetadata], /
    ) -> "Parser[A]":
        return replace(self, formatting=f(self.formatting))

    def map[B](
        self,
        f: Callable[[A], B | dp.ParseError],
        /,
        *,
        catch_exn: bool = False,
    ) -> "Parser[B]":
        """
        Apply a function to the parser's output.

        Arguments:
            f: The function to apply, which is allowed to raise or
                return `ParseError`.
            catch_exn: If `True`, any other exception raised by `f` is
                caught and wrapped into a `ParseError`.
        """

        def parse(ans: dp.Answer) -> B:
            res = self.parse(ans)
            try:
                ret = f(res)
            except dp.ParseError as e:
                raise e
            except Exception as e:
                if catch_exn:
                    raise dp.ParseError(description=str(e))
                else:
                    raise e
            if isinstance(ret, dp.ParseError):
                raise ret
            return ret

        return Parser(
            settings=self.settings, formatting=self.formatting, parse=parse
        )

    def validate(
        self,
        f: Callable[[A], dp.ParseError | None],
        /,
        *,
        catch_exn: bool = False,
    ) -> "Parser[A]":
        """
        Check that the parser's output satisfies a given property.

        If the property is satisfied, function `f` must return `None`.
        Otherwise, it may return or raise a `ParseError`.
        """

        def parse(ans: dp.Answer) -> A:
            res = self.parse(ans)
            try:
                opt_err = f(res)
            except dp.ParseError as e:
                raise e
            except Exception as e:
                if catch_exn:
                    raise dp.ParseError(description=str(e))
                else:
                    raise e
            if opt_err:
                raise opt_err
            return res

        return Parser(
            settings=self.settings, formatting=self.formatting, parse=parse
        )

    @property
    def wrap_errors(self) -> "Parser[A | WrappedParseError]":
        """
        Wrap parse errors into `WrappedParseError`.
        """

        def parse(ans: dp.Answer) -> A | WrappedParseError:
            try:
                return self.parse(ans)
            except dp.ParseError as e:
                return WrappedParseError(e)

        return Parser(
            settings=self.settings, formatting=self.formatting, parse=parse
        )

    def response_with[T: md.AbstractTool[Any]](
        self, tools: TypeAnnot[T]
    ) -> "Parser[Response[A, T]]":
        """
        Wrap answers into full `Response` objects.
        """

        tools_raw = dpi.union_components(tools)
        tools_types: list[type[md.AbstractTool[Any]]] = [
            a for a in tools_raw if issubclass(a, md.AbstractTool)
        ]
        assert len(tools_types) == len(tools_raw), (
            f"Invalid tools union: {tools}"
        )
        if self.settings.tools is None:
            tools_settings = dp.ToolSettings(
                tool_types=tools_types, force_tool_call=False
            )
        else:
            tools_settings = dp.ToolSettings(
                tool_types=[*tools_types, *self.settings.tools.tool_types],
                force_tool_call=self.settings.tools.force_tool_call,
            )
        settings = replace(self.settings, tools=tools_settings)

        def parse(ans: dp.Answer) -> Response[A, T]:
            # If the answer is one of the provided tool types, we
            # return. Otherwise, we call the parser recursively.
            tcs: list[T] = []
            for tc in ans.tool_calls:
                for t in tools_types:
                    if tc.name == t.tool_name():
                        tcs.append(_parse_or_raise(t, tc.args))
                        break
            if tcs:
                return Response[A, T](ans, ToolRequests(tcs))
            else:
                parsed = self.parse(ans)
                return Response[A, T](ans, FinalAnswer(parsed))

        return Parser(
            settings=settings, formatting=self.formatting, parse=parse
        )

    @property
    def response(self) -> "GenericParser":
        """
        Wrap answers into full `Response` objects.

        Return a `GenericParser` so that the list of supported tools can
        be extracted from the query's answer type.
        """

        def parser(annot: TypeAnnot[Any], /) -> Parser[Any]:
            assert typing.get_origin(annot) is Response, (
                f"Response type expected: {annot}"
            )
            args = typing.get_args(annot)
            assert len(args) == 2
            return self.response_with(args[1])

        return GenericParser(parser)

    @property
    def trim(self: "Parser[str]") -> "Parser[str]":
        """
        Trim the output of a string parser.
        """
        return self.map(str.strip)

    @property
    def json(self: "Parser[str]") -> "GenericParser":
        """
        Parse a string as a JSON object.

        Return a `GenericParser` so that the target type can be
        extracted from the query's answer type.
        """

        return GenericParser(self.json_as)

    @property
    def yaml(self: "Parser[str]") -> "GenericParser":
        """
        Parse a string as a YAML object.

        Return a `GenericParser` so that the target type can be
        extracted from the query's answer type.
        """

        return GenericParser(self.yaml_as)

    def json_as[U](self: "Parser[str]", type: TypeAnnot[U]) -> "Parser[U]":
        """
        Parse a string as a JSON object.

        !!! info
            This method currently does not work very well with type
            inference since its arguments do not allow inferring the
            type of `U`. This should work better once `TypeAnnot` can be
            replaced with `TypeExpr` (incoming in Python 3.14).
        """

        _assert_not_response_type(type, where="json_as")
        schema = md.Schema.make(type)
        return self.map(partial(_parse_json_as, type)).update_formatting(
            lambda f: replace(f, what="json", schema=schema)
        )

    def yaml_as[U](self: "Parser[str]", type: TypeAnnot[U]) -> "Parser[U]":
        """
        Parse a string as a YAML object.

        !!! info
            This method currently does not work very well with type
            inference since its arguments do not allow inferring the
            type of `U`. This should work better once `TypeAnnot` can be
            replaced with `TypeExpr` (incoming in Python 3.14).
        """
        _assert_not_response_type(type, where="yaml_as")
        schema = md.Schema.make(type)
        return self.map(partial(_parse_yaml_as, type)).update_formatting(
            lambda f: replace(f, what="yaml", schema=schema)
        )

wrap_errors property

wrap_errors: Parser[A | WrappedParseError]

Wrap parse errors into WrappedParseError.

response property

response: GenericParser

Wrap answers into full Response objects.

Return a GenericParser so that the list of supported tools can be extracted from the query's answer type.

trim property

trim: Parser[str]

Trim the output of a string parser.

json property

json: GenericParser

Parse a string as a JSON object.

Return a GenericParser so that the target type can be extracted from the query's answer type.

yaml property

yaml: GenericParser

Parse a string as a YAML object.

Return a GenericParser so that the target type can be extracted from the query's answer type.

map

map(f: Callable[[A], B | ParseError], /, *, catch_exn: bool = False) -> Parser[B]

Apply a function to the parser's output.

Parameters:

Name	Type	Description	Default
f	Callable[[A], B | ParseError]	The function to apply, which is allowed to raise or return ParseError.	required
catch_exn	bool	If True, any other exception raised by f is caught and wrapped into a ParseError.	False

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

def map[B](
    self,
    f: Callable[[A], B | dp.ParseError],
    /,
    *,
    catch_exn: bool = False,
) -> "Parser[B]":
    """
    Apply a function to the parser's output.

    Arguments:
        f: The function to apply, which is allowed to raise or
            return `ParseError`.
        catch_exn: If `True`, any other exception raised by `f` is
            caught and wrapped into a `ParseError`.
    """

    def parse(ans: dp.Answer) -> B:
        res = self.parse(ans)
        try:
            ret = f(res)
        except dp.ParseError as e:
            raise e
        except Exception as e:
            if catch_exn:
                raise dp.ParseError(description=str(e))
            else:
                raise e
        if isinstance(ret, dp.ParseError):
            raise ret
        return ret

    return Parser(
        settings=self.settings, formatting=self.formatting, parse=parse
    )

validate

validate(
    f: Callable[[A], ParseError | None], /, *, catch_exn: bool = False
) -> Parser[A]

Check that the parser's output satisfies a given property.

If the property is satisfied, function f must return None. Otherwise, it may return or raise a ParseError.

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

def validate(
    self,
    f: Callable[[A], dp.ParseError | None],
    /,
    *,
    catch_exn: bool = False,
) -> "Parser[A]":
    """
    Check that the parser's output satisfies a given property.

    If the property is satisfied, function `f` must return `None`.
    Otherwise, it may return or raise a `ParseError`.
    """

    def parse(ans: dp.Answer) -> A:
        res = self.parse(ans)
        try:
            opt_err = f(res)
        except dp.ParseError as e:
            raise e
        except Exception as e:
            if catch_exn:
                raise dp.ParseError(description=str(e))
            else:
                raise e
        if opt_err:
            raise opt_err
        return res

    return Parser(
        settings=self.settings, formatting=self.formatting, parse=parse
    )

response_with

response_with(tools: TypeAnnot[T]) -> Parser[Response[A, T]]

Wrap answers into full Response objects.

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

def response_with[T: md.AbstractTool[Any]](
    self, tools: TypeAnnot[T]
) -> "Parser[Response[A, T]]":
    """
    Wrap answers into full `Response` objects.
    """

    tools_raw = dpi.union_components(tools)
    tools_types: list[type[md.AbstractTool[Any]]] = [
        a for a in tools_raw if issubclass(a, md.AbstractTool)
    ]
    assert len(tools_types) == len(tools_raw), (
        f"Invalid tools union: {tools}"
    )
    if self.settings.tools is None:
        tools_settings = dp.ToolSettings(
            tool_types=tools_types, force_tool_call=False
        )
    else:
        tools_settings = dp.ToolSettings(
            tool_types=[*tools_types, *self.settings.tools.tool_types],
            force_tool_call=self.settings.tools.force_tool_call,
        )
    settings = replace(self.settings, tools=tools_settings)

    def parse(ans: dp.Answer) -> Response[A, T]:
        # If the answer is one of the provided tool types, we
        # return. Otherwise, we call the parser recursively.
        tcs: list[T] = []
        for tc in ans.tool_calls:
            for t in tools_types:
                if tc.name == t.tool_name():
                    tcs.append(_parse_or_raise(t, tc.args))
                    break
        if tcs:
            return Response[A, T](ans, ToolRequests(tcs))
        else:
            parsed = self.parse(ans)
            return Response[A, T](ans, FinalAnswer(parsed))

    return Parser(
        settings=settings, formatting=self.formatting, parse=parse
    )

json_as

json_as(type: TypeAnnot[U]) -> Parser[U]

Parse a string as a JSON object.

Info

This method currently does not work very well with type inference since its arguments do not allow inferring the type of U. This should work better once TypeAnnot can be replaced with TypeExpr (incoming in Python 3.14).

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

def json_as[U](self: "Parser[str]", type: TypeAnnot[U]) -> "Parser[U]":
    """
    Parse a string as a JSON object.

    !!! info
        This method currently does not work very well with type
        inference since its arguments do not allow inferring the
        type of `U`. This should work better once `TypeAnnot` can be
        replaced with `TypeExpr` (incoming in Python 3.14).
    """

    _assert_not_response_type(type, where="json_as")
    schema = md.Schema.make(type)
    return self.map(partial(_parse_json_as, type)).update_formatting(
        lambda f: replace(f, what="json", schema=schema)
    )

yaml_as

yaml_as(type: TypeAnnot[U]) -> Parser[U]

Parse a string as a YAML object.

Info

This method currently does not work very well with type inference since its arguments do not allow inferring the type of U. This should work better once TypeAnnot can be replaced with TypeExpr (incoming in Python 3.14).

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

def yaml_as[U](self: "Parser[str]", type: TypeAnnot[U]) -> "Parser[U]":
    """
    Parse a string as a YAML object.

    !!! info
        This method currently does not work very well with type
        inference since its arguments do not allow inferring the
        type of `U`. This should work better once `TypeAnnot` can be
        replaced with `TypeExpr` (incoming in Python 3.14).
    """
    _assert_not_response_type(type, where="yaml_as")
    schema = md.Schema.make(type)
    return self.map(partial(_parse_yaml_as, type)).update_formatting(
        lambda f: replace(f, what="yaml", schema=schema)
    )

GenericParser dataclass

A mapping from a query's answer type to a parser specification.

This is useful to avoid redundancy when specifying parsers. In particular, it allows writing:

@dataclass
class MyQuery(Query[Response[Out, Tool1 | Tool2]]):
    ...
    __parser__ = last_block.yaml.response

instead of:

__parser__ = last_block.yaml_as(Out).response_with(Tool1 | Tool2)

Attributes:

Name	Type	Description
for_type	_GenericParserFn	A function that takes a type annotation and returns a Parser for this type.

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

@dataclass(frozen=True)
class GenericParser:
    """
    A mapping from a query's answer type to a parser specification.

    This is useful to avoid redundancy when specifying parsers. In
    particular, it allows writing:

    ```python
    @dataclass
    class MyQuery(Query[Response[Out, Tool1 | Tool2]]):
        ...
        __parser__ = last_block.yaml.response
    ```

    instead of:

    ```python
    __parser__ = last_block.yaml_as(Out).response_with(Tool1 | Tool2)
    ```

    Attributes:
        for_type: A function that takes a type annotation and returns a
            `Parser` for this type.
    """

    for_type: "_GenericParserFn"

    @property
    def wrap_errors(self) -> "GenericParser":
        """
        Wrap parse errors into `WrappedParseError`.

        A runtime check is performed to ensure that the answer type
        features `WrappedParseError`.
        """

        def parser(annot: TypeAnnot[Any], /) -> Parser[Any]:
            comps = dpi.union_components(annot)
            assert len(comps) >= 2 and WrappedParseError in comps, (
                "Answer type does not have shape `... | WrappedParseError`: "
                + f"{annot}"
            )
            annot = dpi.make_union(
                [c for c in comps if c != WrappedParseError]
            )
            return self.for_type(annot).wrap_errors

        return GenericParser(parser)

    @property
    def response(self) -> "GenericParser":
        """
        Wrap answers into full `Response` objects.

        Possible tool calls are extracted from the query's answer type
        and an exception is raised if this type does not have the form
        `Response[..., ...]`.
        """

        def parser(annot: TypeAnnot[Any], /) -> Parser[Any]:
            assert typing.get_origin(annot) is Response, (
                f"Response type expected: {annot}"
            )
            args = typing.get_args(annot)
            assert len(args) == 2
            return self.for_type(args[0]).response_with(args[1])

        return GenericParser(parser)

wrap_errors property

wrap_errors: GenericParser

Wrap parse errors into WrappedParseError.

A runtime check is performed to ensure that the answer type features WrappedParseError.

response property

response: GenericParser

Wrap answers into full Response objects.

Possible tool calls are extracted from the query's answer type and an exception is raised if this type does not have the form Response[..., ...].

_GenericParserFn

Bases: Protocol

Type of functions wrapped by GenericParser.

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

class _GenericParserFn(Protocol):
    """
    Type of functions wrapped by `GenericParser`.
    """

    def __call__[T](self, type: TypeAnnot[T], /) -> Parser[T]: ...

ParserDict

ParserDict: dict[AnswerMode, Parser[Any] | GenericParser]

A mapping from answer modes to parser specifications.

Can be used as a value for the __parser__ class attribute of queries.

Response dataclass

Answer type for queries that allow follow-ups.

Response values give access to both the raw LLM response (to be passed pass in AnswerPrefix) and to eventual tool calls. See the Parser.response, Parser.response_with, and GenericParser.response methods for creating parsers that produce Response values.

Attributes:

Name	Type	Description
answer	Answer	The raw, unparsed LLM answer.
parsed	FinalAnswer[F] | ToolRequests[T]	Either the parsed answer wrapped in FinalAnswer or some tool call requests wrapped in ToolRequests.

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

@dataclass(frozen=True)
class Response[F, T: md.AbstractTool[Any]]:
    """
    Answer type for queries that allow follow-ups.

    `Response` values give access to both the raw LLM response (to be
    passed pass in `AnswerPrefix`) and to eventual tool calls. See the
    `Parser.response`, `Parser.response_with`, and
    `GenericParser.response` methods for creating parsers that produce
    `Response` values.

    Attributes:
        answer: The raw, unparsed LLM answer.
        parsed: Either the parsed answer wrapped in `FinalAnswer` or
            some tool call requests wrapped in `ToolRequests`.
    """

    answer: dp.Answer
    parsed: FinalAnswer[F] | ToolRequests[T]

FinalAnswer dataclass

See Response.

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

@dataclass(frozen=True)
class FinalAnswer[F]:
    """
    See `Response`.
    """

    final: F

ToolRequests dataclass

See Response.

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

@dataclass(frozen=True)
class ToolRequests[T: md.AbstractTool[Any]]:
    """
    See `Response`.
    """

    tool_calls: Sequence[T]

WrappedParseError dataclass

A wrapped parse error that is returned to a strategy instead of causing a failure.

For queries that declare a return type of the form Response[... | WrappedParseError, ...], parse errors do not result in failures but are instead wrapped and returned, to be handled explicitly by the surrounding strategy. For example, when building conversational agents with interact, having the query include WrappedParseError in its return type allows explicitly asking the agent to fix parse errors instead of failing (or having the policy retry an identical prompt).

See the Parser.wrap_errors and GenericParser.wrap_errors methods for creating parsers that produce WrappedParseError values.

Attributes:

Name	Type	Description
error	ParseError	The wrapped parse error.

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

@dataclass
class WrappedParseError:
    """
    A wrapped parse error that is returned to a strategy instead of
    causing a failure.

    For queries that declare a return type of the form `Response[... |
    WrappedParseError, ...]`, parse errors do not result in failures but
    are instead wrapped and returned, to be handled explicitly by the
    surrounding strategy. For example, when building conversational
    agents with `interact`, having the query include `WrappedParseError`
    in its return type allows explicitly asking the agent to fix parse
    errors instead of failing (or having the policy retry an identical
    prompt).

    See the `Parser.wrap_errors` and `GenericParser.wrap_errors` methods
    for creating parsers that produce `WrappedParseError` values.

    Attributes:
        error: The wrapped parse error.
    """

    error: dp.ParseError

QueryTemplateArgs

Bases: TypedDict

Template arguments passed to all query templates.

For particular kinds of templates, additional arguments may be provided (e.g., feedback for feedback prompts).

Attributes:

Name	Type	Description
query	Query[Any]	The query instance.
mode	AnswerMode	The requested answer mode.
available_modes	Sequence[AnswerMode]	The sequence of all available answer modes for the query type.
params	dict[str, Any]	The query hyperparameters (e.g., as passed to few_shot)
format	FormattingMetadata	Formatting metadata.

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

class QueryTemplateArgs(typing.TypedDict):
    """
    Template arguments passed to all query templates.

    For particular kinds of templates, additional arguments may be
    provided (e.g., `feedback` for feedback prompts).

    Attributes:
        query: The query instance.
        mode: The requested answer mode.
        available_modes: The sequence of all available answer modes for
            the query type.
        params: The query hyperparameters (e.g., as passed to `few_shot`)
        format: Formatting metadata.
    """

    # TODO: in future Python versions, use `extra_items=Any` (PEP 728)

    query: "Query[Any]"
    mode: dp.AnswerMode
    available_modes: Sequence[dp.AnswerMode]
    params: dict[str, Any]
    format: FormattingMetadata

Standard Parsers

structured_as

structured_as(type: TypeAnnot[T]) -> Parser[T]

Parse an LLM structured answer into a given target type.

Warning

Only dataclass types are supported, since most LLM providers only support structured output and tool calls for JSON objects.

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

def structured_as[T](type: TypeAnnot[T], /) -> Parser[T]:
    """
    Parse an LLM structured answer into a given target type.

    !!! warning
    Only dataclass types are supported, since most LLM providers
    only support structured output and tool calls for JSON objects.
    """
    _assert_not_response_type(type, where="structured_as")
    _check_valid_structured_output_type(type)
    settings = dp.QuerySettings(dp.StructuredOutputSettings(type))
    formatting = FormattingMetadata(
        where="full_answer", what="json", schema=md.Schema.make(type)
    )
    return Parser(
        settings, formatting, lambda ans: _parse_structured_output(type, ans)
    )

final_tool_call_as

final_tool_call_as(annot: TypeAnnot[T]) -> Parser[T]

Variant of structured_as, where the query answer type is presented to oracles as a tool, which must be called to produce the final answer. This provides an alternative to "structured", which additionally allows a chain of thoughts to precede the final answer.

Warning

Only dataclass types are supported, since most LLM providers only support structured output and tool calls for JSON objects.

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

def final_tool_call_as[T](annot: TypeAnnot[T], /) -> Parser[T]:
    """
    Variant of `structured_as`, where the query answer type is presented
    to oracles as a tool, which must be called to produce the final
    answer. This provides an alternative to "structured", which
    additionally allows a chain of thoughts to precede the final answer.

    !!! warning
        Only dataclass types are supported, since most LLM providers
        only support structured output and tool calls for JSON objects.
    """
    _check_valid_structured_output_type(annot)
    assert isinstance(annot, type)  # redundant with previous check
    tool = cast(type[Any], annot)
    tool_settings = dp.ToolSettings(tool_types=[tool], force_tool_call=True)
    settings = dp.QuerySettings(None, tool_settings)
    formatting = FormattingMetadata(
        where="tool_call", what="json", schema=md.Schema.make(annot)
    )

    def parse(ans: dp.Answer) -> T:
        assert len(ans.tool_calls) == 1, (
            f"Expected one final tool call, got answer: {ans}"
        )
        return _parse_or_raise(tool, ans.tool_calls[0].args)

    return Parser(settings, formatting, parse)

structured module-attribute

structured = GenericParser(structured_as)

Generic parser associated with structured_as.

final_tool_call module-attribute

final_tool_call = GenericParser(final_tool_call_as)

Generic parser associated with final_tool_call_as.

last_code_block module-attribute

last_code_block: Parser[str] = update_formatting(
    lambda f: replace(f, where="last_code_block")
)

Parser that extracts the last code block from a text answer.

get_text module-attribute

get_text = Parser[str](
    QuerySettings(),
    FormattingMetadata(where="full_answer", what="text"),
    _get_text_answer,
)

Parser that extracts the text content of an answer.

A runtime error is raised if the answer contains structured content.

Prompting Policies

few_shot

few_shot(
    query: AttachedQuery[T],
    env: PolicyEnv,
    model: LLM,
    *,
    params: dict[str, object] | None = None,
    select_examples: Sequence[ExampleSelector] = (),
    mode: AnswerMode = None,
    enable_logging: bool = True,
    temperature: float | None = None,
    num_completions: int = 1,
    max_requests: int | None = None,
    no_wrap_parse_errors: bool = False,
    iterative_mode: bool = False,
) -> StreamGen[T]

The standard few-shot prompting policy.

A prompt is formed by concatenating a system prompt, a series of examples (each of which consists in an instance prompt followed by an answer), and a final answer prompt. Then, answers are repeatedly sampled and parsed, until a spending request is declined.

Parameters:

Name	Type	Description	Default
query	AttachedQuery[T]	The query to answer. env: The policy environment.	required
model	LLM	The LLM to use for answering the query	required
params	dict[str, object] | None	Prompt hyperparameters, which are passed to prompt templates as a params dictionary.	None
select_examples	Sequence[ExampleSelector]	A series of filters for selecting examples, to be applied in sequence. By default, no filter is used and so all available examples are fetched.	()
mode	AnswerMode	The answer mode to use for parsing the query answer.	None
enable_logging	bool	Whether to log raw oracle responses.	True
temperature	float | None	The temperature parameter to use with the LLM, as a number from 0 to 2.	None
num_completions	int	The number of completions to request for each LLM call. Note that most LLM providers only bill input tokens once, regardless of the number of completions.	1
max_requests	int | None	The maximum number of LLM requests to perform before the resulting seach stream terminates, if any.	None
no_wrap_parse_errors	bool	If set to True, then parser results of type WrappedParseError are unwrapped and treated as normal parse errors.	False
iterative_mode	bool	If set to False (default), answers are repeatedly and independently sampled. If set to True, a single chat conversation occurs instead: Whenever a parse error occurs, a message is issued by rendering the <QueryName>.repair.jinja template, asking for a new attempt to be made (the ParseError object is available as an error template variable). After an answer is successfully generated and parsed, a message is issued by rendering the <QueryName>.more.jinja template, asking for another different answer to be generated. This special mode allows creating simple conversational agents with very little effort, by only defining a single query. However, it does not support tool calls, and the demonstration language cannot be used to illustrate how repair and more messages should be handled. For implementing more advanced conversational agents, see the standard interact strategy.	False

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

@prompting_policy
def few_shot[T](
    query: dp.AttachedQuery[T],
    env: PolicyEnv,
    model: md.LLM,
    *,
    params: dict[str, object] | None = None,
    select_examples: Sequence[ExampleSelector] = (),
    mode: dp.AnswerMode = None,
    enable_logging: bool = True,
    temperature: float | None = None,
    num_completions: int = 1,
    max_requests: int | None = None,
    no_wrap_parse_errors: bool = False,
    iterative_mode: bool = False,
) -> dp.StreamGen[T]:
    """
    The standard few-shot prompting policy.

    A prompt is formed by concatenating a system prompt, a series of
    examples (each of which consists in an instance prompt followed by
    an answer), and a final answer prompt. Then, answers are repeatedly
    sampled and parsed, until a spending request is declined.

    Arguments:
        query: The query to answer. env: The policy environment.
        model: The LLM to use for answering the query
        params: Prompt hyperparameters, which are passed to prompt
            templates as a `params` dictionary.
        select_examples: A series of filters for selecting examples, to
            be applied in sequence. By default, no filter is used and so
            all available examples are fetched.
        mode: The answer mode to use for parsing the query answer.
        enable_logging: Whether to log raw oracle responses.
        temperature: The temperature parameter to use with the LLM, as a
            number from 0 to 2.
        num_completions: The number of completions to request for each
            LLM call. Note that most LLM providers only bill input
            tokens once, regardless of the number of completions.
        max_requests: The maximum number of LLM requests to perform
            before the resulting seach stream terminates, if any.
        no_wrap_parse_errors: If set to `True`, then parser results of
            type `WrappedParseError` are unwrapped and treated as normal
            parse errors.
        iterative_mode: If set to `False` (default), answers are
            repeatedly and independently sampled. If set to `True`, a
            single chat conversation occurs instead: Whenever a parse
            error occurs, a message is issued by rendering the
            `<QueryName>.repair.jinja` template, asking for a new
            attempt to be made (the `ParseError` object is available as
            an `error` template variable). After an answer is
            successfully generated and parsed, a message is issued by
            rendering the `<QueryName>.more.jinja` template, asking for
            another different answer to be generated.

            This special mode allows creating simple conversational
            agents with very little effort, by only defining a single
            query. However, it does not support tool calls, and the
            demonstration language cannot be used to illustrate how
            `repair` and `more` messages should be handled. For
            implementing more advanced conversational agents, see
            the standard `interact` strategy.
    """
    assert not iterative_mode or num_completions == 1
    assert max_requests is None or max_requests > 0
    env.tracer.trace_query(query.ref)
    examples = fetch_examples(env.examples, query.query, select_examples)
    mngr = env.templates
    if params is None:
        params = {}
    prompt = create_prompt(query.query, examples, params, mode, mngr)
    settings = query.query.query_settings(mode)
    options: md.RequestOptions = {}
    if temperature is not None:
        options["temperature"] = temperature
    structured_output = None
    if settings.structured_output is not None:
        out_type = settings.structured_output.type
        structured_output = md.Schema.make(out_type)
    tools = []
    if settings.tools is not None:
        if settings.tools.force_tool_call:
            options["tool_choice"] = "required"
        tools = [md.Schema.make(t) for t in settings.tools.tool_types]
    num_reqs = 0
    while max_requests is None or num_reqs < max_requests:
        num_reqs += 1
        req = md.LLMRequest(
            prompt,
            num_completions=num_completions,
            options=options,
            tools=tools,
            structured_output=structured_output,
        )
        resp = yield from _send_request(model, req, env)
        if isinstance(resp, SpendingDeclined):
            return
        log_oracle_response(env, query, req, resp, verbose=enable_logging)
        if not resp.outputs:
            log(env, "llm_no_output", loc=query)
            continue
        elements: list[dp.Tracked[T] | dp.ParseError] = []
        answers: list[dp.Answer] = []
        for output in resp.outputs:
            answer = dp.Answer(mode, output.content, tuple(output.tool_calls))
            answers.append(answer)
            element = query.parse_answer(answer)
            if no_wrap_parse_errors:
                element = _unwrap_parse_error(element)
            env.tracer.trace_answer(query.ref, answer)
            if isinstance(element, dp.ParseError):
                log(env, "parse_error", {"error": element}, loc=query)
            elements.append(element)
        for element in elements:
            if not isinstance(element, dp.ParseError):
                yield dp.Solution(element)
        # In iterative mode, we want to keep the conversation going
        if iterative_mode:
            assert len(elements) == 1 and len(answers) == 1
            element = elements[0]
            if isinstance(element, dp.ParseError):
                try:
                    repair = query.query.generate_prompt(
                        kind=REPAIR_PROMPT,
                        mode=mode,
                        params={"params": params, "error": element},
                        env=mngr,
                    )
                except dp.TemplateFileMissing:
                    repair = (
                        "Invalid answer. Please consider the following"
                        + f" feedback and try again:\n\n{element}"
                    )
                new_message = md.UserMessage(repair)
            else:
                try:
                    gen_new = query.query.generate_prompt(
                        kind=REQUEST_OTHER_PROMPT,
                        mode=mode,
                        params={"params": params},
                        env=mngr,
                    )
                except dp.TemplateFileMissing:
                    gen_new = "Good! Can you generate a different answer now?"
                new_message = md.UserMessage(gen_new)

            prompt = (*prompt, md.AssistantMessage(answers[0]), new_message)

classify

classify(
    query: AttachedQuery[T],
    env: PolicyEnv,
    model: LLM,
    params: dict[str, object] | None = None,
    select_examples: Sequence[ExampleSelector] = (),
    mode: AnswerMode = None,
    enable_logging: bool = True,
    top_logprobs: int = 20,
    temperature: float = 1.0,
    bias: tuple[str, float] | None = None,
) -> StreamGen[T]

Execute a classification query, attaching a probability distribution to the attached answer.

Parameters:

Name	Type	Description	Default
query	AttachedQuery[T]	The query to answer.	required
env	PolicyEnv	The global policy environment.	required
model	LLM	The LLM to use for answering the query.	required
params	dict[str, object] | None	Prompt hyperparameters.	None
select_examples	Sequence[ExampleSelector]	Example selector.	()
mode	AnswerMode	The answer mode to use for parsing the query answer.	None
enable_logging	bool	Whether to log raw oracle responses.	True
top_logprobs	int	The number of top logprobs to request from the LLM, putting an upper bound on the support size of the classifier's output distributions.	20
temperature	float	A temperature to apply to the classifier's output distribution (a temperature of 0 means that only top elements are assigned a nonzero probability).	1.0
bias	tuple[str, float] | None	When bias=(e, p) is provided, the final classifier distribution D is transformed into (1-p)*D + p*dirac(e)	None

See few_shot for details on some of the arguments above.

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

@prompting_policy
def classify[T](
    query: dp.AttachedQuery[T],
    env: PolicyEnv,
    model: md.LLM,
    params: dict[str, object] | None = None,
    select_examples: Sequence[ExampleSelector] = (),
    mode: dp.AnswerMode = None,
    enable_logging: bool = True,
    top_logprobs: int = 20,
    temperature: float = 1.0,
    bias: tuple[str, float] | None = None,
) -> dp.StreamGen[T]:
    """
    Execute a classification query, attaching a probability distribution
    to the attached answer.

    Arguments:
        query: The query to answer.
        env: The global policy environment.
        model: The LLM to use for answering the query.
        params: Prompt hyperparameters.
        select_examples: Example selector.
        mode: The answer mode to use for parsing the query answer.
        enable_logging: Whether to log raw oracle responses.
        top_logprobs: The number of top logprobs to request from the
            LLM, putting an upper bound on the support size of the
            classifier's output distributions.
        temperature: A temperature to apply to the classifier's output
            distribution (a temperature of 0 means that only top
            elements are assigned a nonzero probability).
        bias: When `bias=(e, p)` is provided, the final classifier
            distribution `D` is transformed into `(1-p)*D + p*dirac(e)`

    See `few_shot` for details on some of the arguments above.
    """
    env.tracer.trace_query(query.ref)
    examples = fetch_examples(env.examples, query.query, select_examples)
    mngr = env.templates
    if params is None:
        params = {}
    prompt = create_prompt(query.query, examples, params, mode, mngr)
    aset = query.query.finite_answer_set()
    assert aset is not None
    vals: list[str] = []
    for a in aset:
        assert isinstance(a.content, str)
        vals.append(a.content)
    options: md.RequestOptions = {
        "logprobs": True,
        "top_logprobs": top_logprobs,
        # TODO: somehow, there seems to be a problem with this, where
        # one can get an empty answer with "finish_reason: length":
        # "max_completion_tokens": 1,
        "temperature": 0.0,
    }
    req = md.LLMRequest(
        prompt,
        num_completions=1,
        options=options,
    )
    resp = yield from _send_request(model, req, env)
    if isinstance(resp, SpendingDeclined):
        return
    log_oracle_response(env, query, req, resp, verbose=enable_logging)
    if not resp.outputs:
        return
    output = resp.outputs[0]
    answer = dp.Answer(mode, output.content)
    env.tracer.trace_answer(query.ref, answer)
    parse = partial(_parse_or_log_and_raise, query=query, env=env)
    try:
        element = parse(answer)
        lpinfo = output.logprobs
        assert lpinfo is not None
        ldistr = _compute_value_distribution(vals, lpinfo[0])
        if not ldistr:
            assert isinstance(output.content, str)
            ldistr = {output.content: 0.0}
        distr = _apply_temperature(ldistr, temperature)
        if bias is not None:
            distr = _apply_bias(distr, bias)
        distr_tup = [(parse(dp.Answer(mode, k)), p) for k, p in distr.items()]
        meta = ProbInfo(distr_tup)
        yield dp.Solution(element, meta)
    except dp.ParseError:
        return

ProbInfo dataclass

Bases: SearchMeta

Distribution probability, guaranteed to be nonempty and to sum to 1.

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

@dataclass
class ProbInfo(dp.SearchMeta):
    """
    Distribution probability, guaranteed to be nonempty and to sum to 1.
    """

    distr: Sequence[tuple[dp.Tracked[Any], float]]

Models

LLM

Bases: ABC

Base class for an LLM.

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

class LLM(ABC):
    """
    Base class for an LLM.
    """

    def estimate_budget(self, req: LLMRequest) -> Budget:
        """
        Estimate the budget that is required to process a request.
        """
        return Budget({NUM_REQUESTS: 1, NUM_COMPLETIONS: req.num_completions})

    @abstractmethod
    def add_model_defaults(self, req: LLMRequest) -> LLMRequest:
        """
        Rewrite a request to take model-specific defaults into account.

        A model can carry default values for some of the request fields
        (e.g. the model name). Thus, requests must be processed through
        this function right before they are executed or cached.
        """
        pass

    @abstractmethod
    def _send_final_request(self, req: LLMRequest) -> LLMResponse:
        """
        Core method for processing a request.

        To be overriden by subclasses to implement the core
        functionality of `send_request`. The latter additionally handles
        model=specific defaults and caching.

        This function is allowed to raise exceptions (some
        provider-specific), including `LLMBusyException` for cases where
        retrials may be warranted.
        """
        pass

    def stream_request(
        self, chat: Chat, options: RequestOptions
    ) -> AsyncIterable[str]:
        """
        Stream the text answer to a request.

        This is currently not used but could be leveraged by the VSCode
        extension in the future.
        """
        raise StreamingNotImplemented()

    @final
    def send_request(
        self, req: LLMRequest, cache: "LLMCache | None"
    ) -> LLMResponse:
        """
        Send a request to a model and return the response.

        This function is allowed to raise exceptions (some
        provider-specific), including `LLMBusyException` for cases where
        retrials may be warranted.

        Attributes:
            req: The request to send.
            cache: An optional cache to use for the request.
        """
        if cache is not None:
            self = CachedModel(self, cache)
        full_req = self.add_model_defaults(req)
        return self._send_final_request(full_req)

estimate_budget

estimate_budget(req: LLMRequest) -> Budget

Estimate the budget that is required to process a request.

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

def estimate_budget(self, req: LLMRequest) -> Budget:
    """
    Estimate the budget that is required to process a request.
    """
    return Budget({NUM_REQUESTS: 1, NUM_COMPLETIONS: req.num_completions})

add_model_defaults abstractmethod

add_model_defaults(req: LLMRequest) -> LLMRequest

Rewrite a request to take model-specific defaults into account.

A model can carry default values for some of the request fields (e.g. the model name). Thus, requests must be processed through this function right before they are executed or cached.

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

@abstractmethod
def add_model_defaults(self, req: LLMRequest) -> LLMRequest:
    """
    Rewrite a request to take model-specific defaults into account.

    A model can carry default values for some of the request fields
    (e.g. the model name). Thus, requests must be processed through
    this function right before they are executed or cached.
    """
    pass

stream_request

stream_request(chat: Chat, options: RequestOptions) -> AsyncIterable[str]

Stream the text answer to a request.

This is currently not used but could be leveraged by the VSCode extension in the future.

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

def stream_request(
    self, chat: Chat, options: RequestOptions
) -> AsyncIterable[str]:
    """
    Stream the text answer to a request.

    This is currently not used but could be leveraged by the VSCode
    extension in the future.
    """
    raise StreamingNotImplemented()

send_request

send_request(req: LLMRequest, cache: LLMCache | None) -> LLMResponse

Send a request to a model and return the response.

This function is allowed to raise exceptions (some provider-specific), including LLMBusyException for cases where retrials may be warranted.

Attributes:

Name	Type	Description
req		The request to send.
cache		An optional cache to use for the request.

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

@final
def send_request(
    self, req: LLMRequest, cache: "LLMCache | None"
) -> LLMResponse:
    """
    Send a request to a model and return the response.

    This function is allowed to raise exceptions (some
    provider-specific), including `LLMBusyException` for cases where
    retrials may be warranted.

    Attributes:
        req: The request to send.
        cache: An optional cache to use for the request.
    """
    if cache is not None:
        self = CachedModel(self, cache)
    full_req = self.add_model_defaults(req)
    return self._send_final_request(full_req)

LLMRequest dataclass

An LLM chat completion request.

Attributes:

Name	Type	Description
chat	Chat	The chat history.
num_completions	int	The number of completions to generate. Note that most LLM providers only bill input tokens once, regardless of the number of requested completions.
options	RequestOptions	Request options.
tools	Sequence[Schema]	Available tools.
structured_output	Schema | None	Provide a schema to enable structured output, or None for disabling it.

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

@dataclass(frozen=True)
class LLMRequest:
    """
    An LLM chat completion request.

    Attributes:
        chat: The chat history.
        num_completions: The number of completions to generate. Note
            that most LLM providers only bill input tokens once,
            regardless of the number of requested completions.
        options: Request options.
        tools: Available tools.
        structured_output: Provide a schema to enable structured output,
            or `None` for disabling it.
    """

    chat: Chat
    num_completions: int
    options: RequestOptions
    tools: Sequence[Schema] = ()
    structured_output: Schema | None = None

    def __hash__(self) -> int:
        # LLMRequest needs to be hashable for `CachedModel` to work.
        return hash(repr((self.chat, self.num_completions, self.options)))

LLMResponse dataclass

Response to an LLM request.

Attributes:

Name	Type	Description
outputs	Sequence[LLMOutput]	Generated completions.
budget	Budget	Budget consumed by the request.
log_items	list[LLMResponseLogItem]	Log items generated while evaluating the request.
model_name	str | None	The name of the model used for the request, which is sometimes more detailed than the model name passed in RequestOptions (e.g., gpt-4.1-mini-2025-04-14 vs gpt-4.1-mini).
usage_info	dict[str, Any] | None	Additional usage info metadata, in a provider-specific format.

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

@dataclass
class LLMResponse:
    """
    Response to an LLM request.

    Attributes:
        outputs: Generated completions.
        budget: Budget consumed by the request.
        log_items: Log items generated while evaluating the request.
        model_name: The name of the model used for the request, which is
            sometimes more detailed than the model name passed in
            `RequestOptions` (e.g., `gpt-4.1-mini-2025-04-14` vs
            `gpt-4.1-mini`).
        usage_info: Additional usage info metadata, in a
            provider-specific format.
    """

    outputs: Sequence[LLMOutput]
    budget: Budget
    log_items: list[LLMResponseLogItem]
    model_name: str | None = None
    usage_info: dict[str, Any] | None = None

AbstractTool

Base class for an LLM tool interface.

A new tool interface can be added by defining a dataclass S that inherits AbstractTool[T], with T the tool output type. Instances of S correspond to tool calls, and an actual tool implementation maps values of type S to values of type T.

A JSON tool specification can be extracted through the tool_name, tool_description and tool_answer_type class methods. The render_result method describes how to render the output of a tool implementation, in a way that can be added back as a message in a chat history.

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

class AbstractTool[T]:
    """
    Base class for an LLM tool interface.

    A new tool interface can be added by defining a dataclass `S` that
    inherits `AbstractTool[T]`, with `T` the tool output type. Instances
    of `S` correspond to tool calls, and an actual tool implementation
    maps values of type `S` to values of type `T`.

    A JSON tool specification can be extracted through the
    `tool_name`, `tool_description` and `tool_answer_type` class
    methods. The `render_result` method describes how to render the
    output of a tool implementation, in a way that can be added back as
    a message in a chat history.
    """

    @classmethod
    def tool_name(cls) -> str:
        return tool_name_of_class_name(cls.__name__)

    @classmethod
    def tool_description(cls) -> str | None:
        return inspect.getdoc(cls)

    @classmethod
    def tool_answer_type(cls) -> TypeAnnot[T]:
        return dpi.first_parameter_of_base_class(cls)

    def render_result(self, res: T) -> str | Structured:
        if isinstance(res, str):
            return res
        ans_type = self.tool_answer_type()
        return Structured(pydantic_dump(ans_type, res))

Chat

Chat: Sequence[ChatMessage]

ChatMessage

ChatMessage: SystemMessage | UserMessage | AssistantMessage | ToolMessage

SystemMessage dataclass

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

@dataclass(frozen=True)
class SystemMessage:
    role: Literal["system"]  # for deserialization
    content: str

    def __init__(self, content: str):
        # to bypass the frozen dataclass check
        object.__setattr__(self, "role", "system")
        object.__setattr__(self, "content", content)

UserMessage dataclass

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

@dataclass(frozen=True)
class UserMessage:
    role: Literal["user"]
    content: str

    def __init__(self, content: str):
        object.__setattr__(self, "role", "user")
        object.__setattr__(self, "content", content)

AssistantMessage dataclass

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

@dataclass(frozen=True)
class AssistantMessage:
    role: Literal["assistant"]
    answer: Answer  # Note: the mode does not really matter for the LLM.

    def __init__(self, answer: Answer):
        # to bypass the frozen dataclass check
        object.__setattr__(self, "role", "assistant")
        object.__setattr__(self, "answer", answer)

ToolMessage dataclass

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

@dataclass(frozen=True)
class ToolMessage:
    role: Literal["tool"]
    call: ToolCall
    result: str | Structured

    def __init__(self, call: ToolCall, result: str | Structured):
        object.__setattr__(self, "role", "tool")
        object.__setattr__(self, "call", call)
        object.__setattr__(self, "result", result)

RequestOptions

Bases: TypedDict

LLM request options, inspired from the OpenAI chat API.

All values are optional.

Attributes:

Name	Type	Description
model	str	The name of the model to use for the request.
reasoning_effort	Literal['minimal', 'low', 'medium', 'high']	The reasoning effort to use for the request, when applicable (e.g., for GPT-5 or o3).
tool_choice	Literal['auto', 'none', 'required']	How the model should select which tool (or tools) to use when generating a response. none means the model will not call any tool and instead generates a message. auto means the model can pick between generating a message or calling one or more tools. required means the model must call one or more tools.
temperature	float	The temperature to use for sampling, as a value between 0 and 2.
max_completion_tokens	int	The maximum number of tokens to generate.
logprobs	bool	Whether to return log probabilities for the generated tokens.
top_logprobs	int	The number of top log probabilities to return for each generated token, as an integer between 0 and 20.

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

class RequestOptions(typing.TypedDict, total=False):
    """
    LLM request options, inspired from the OpenAI chat API.

    All values are optional.

    Attributes:
        model: The name of the model to use for the request.
        reasoning_effort: The reasoning effort to use for the request,
            when applicable (e.g., for GPT-5 or o3).
        tool_choice: How the model should select which tool (or tools)
            to use when generating a response. `none` means the model
            will not call any tool and instead generates a message.
            `auto` means the model can pick between generating a message
            or calling one or more tools. `required` means the model must
            call one or more tools.
        temperature: The temperature to use for sampling, as a value
            between 0 and 2.
        max_completion_tokens: The maximum number of tokens to generate.
        logprobs: Whether to return log probabilities for the generated
            tokens.
        top_logprobs: The number of top log probabilities to return for
            each generated token, as an integer between 0 and 20.
    """

    model: str
    reasoning_effort: Literal["minimal", "low", "medium", "high"]
    tool_choice: Literal["auto", "none", "required"]
    temperature: float
    max_completion_tokens: int
    logprobs: bool
    top_logprobs: int  # from 0 to 20

Schema dataclass

The description of a schema for structured output or tool use.

Attributes:

Name	Type	Description
name	str	Name of the tool or structured output type.
description	str | None	Optional description.
schema	Any	The JSON schema of the tool or structured output type, typically generated using pydantic's json_schema method.

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

@dataclass(frozen=True)
class Schema:
    """
    The description of a schema for structured output or tool use.

    Attributes:
        name: Name of the tool or structured output type.
        description: Optional description.
        schema: The JSON schema of the tool or structured output type,
            typically generated using pydantic's `json_schema` method.
    """

    name: str
    description: str | None
    schema: Any

    @staticmethod
    def make(annot: TypeAnnot[Any], /) -> "Schema":
        """
        Build a schema from a Python type annotation
        """
        if isinstance(annot, type):
            if issubclass(annot, AbstractTool):
                name = annot.tool_name()
                description = annot.tool_description()
            else:
                name = tool_name_of_class_name(annot.__name__)
                # For a dataclass, if no docstring is provided,
                # `inspect.getdoc` shows its signature (name, attribute
                # names and types).
                description = inspect.getdoc(cast(Any, annot))
        elif isinstance(annot, typing.TypeAliasType):
            # TODO: we can do better here.
            name = str(annot)
            description = None
        else:
            # Any other type annotation, such as a union.
            name = str(annot)
            description = None
        adapter = pydantic.TypeAdapter(cast(Any, annot))
        return Schema(
            name=name,
            description=description,
            schema=adapter.json_schema(),
        )

make staticmethod

make(annot: TypeAnnot[Any]) -> Schema

Build a schema from a Python type annotation

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

@staticmethod
def make(annot: TypeAnnot[Any], /) -> "Schema":
    """
    Build a schema from a Python type annotation
    """
    if isinstance(annot, type):
        if issubclass(annot, AbstractTool):
            name = annot.tool_name()
            description = annot.tool_description()
        else:
            name = tool_name_of_class_name(annot.__name__)
            # For a dataclass, if no docstring is provided,
            # `inspect.getdoc` shows its signature (name, attribute
            # names and types).
            description = inspect.getdoc(cast(Any, annot))
    elif isinstance(annot, typing.TypeAliasType):
        # TODO: we can do better here.
        name = str(annot)
        description = None
    else:
        # Any other type annotation, such as a union.
        name = str(annot)
        description = None
    adapter = pydantic.TypeAdapter(cast(Any, annot))
    return Schema(
        name=name,
        description=description,
        schema=adapter.json_schema(),
    )

LLMOutput dataclass

A single LLM chat completion.

Attributes:

Name	Type	Description
content	str | Structured	The completion content, as a string or as a structured object (if structured output was requested).
tool_calls	Sequence[ToolCall]	A sequence of tool calls made by the model, if any.
finish_reason	FinishReason	The reason why the model stopped generating content.
logprobs	Sequence[TokenInfo] | None	Optional sequence of token log probabilities, if requested.
reasoning_content	str | None	Reasoning chain of thoughts, if provided (the DeepSeek API returns reasoning tokens, while the OpenAI API generally does not).

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

@dataclass
class LLMOutput:
    """
    A single LLM chat completion.

    Attributes:
        content: The completion content, as a string or as a structured
            object (if structured output was requested).
        tool_calls: A sequence of tool calls made by the model, if any.
        finish_reason: The reason why the model stopped generating
            content.
        logprobs: Optional sequence of token log probabilities, if
            requested.
        reasoning_content: Reasoning chain of thoughts, if provided (the
            DeepSeek API returns reasoning tokens, while the OpenAI API
            generally does not).
    """

    content: str | Structured
    tool_calls: Sequence[ToolCall]
    finish_reason: FinishReason
    logprobs: Sequence[TokenInfo] | None = None
    reasoning_content: str | None = None

FinishReason

FinishReason: Literal['stop', 'length', 'content_filter', 'tool_calls']

Reason why the LLM stopped generating content.

DummyModel dataclass

Bases: LLM

A model that always fails to generate completions.

Used by the answer_query command in particular.

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

@dataclass
class DummyModel(LLM):
    """
    A model that always fails to generate completions.

    Used by the `answer_query` command in particular.
    """

    @override
    def add_model_defaults(self, req: LLMRequest) -> LLMRequest:
        return req

    @override
    def _send_final_request(self, req: LLMRequest) -> LLMResponse:
        budget = Budget({NUM_REQUESTS: req.num_completions})
        return LLMResponse(
            outputs=[], budget=budget, log_items=[], model_name="<dummy>"
        )

WithRetry dataclass

Bases: LLM

Retrying with exponential backoff.

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

@dataclass
class WithRetry(LLM):
    """
    Retrying with exponential backoff.
    """

    model: LLM
    num_attempts: int = 5
    base_delay_seconds: float = 1.0
    exponential_factor: float = 2.0
    delay_noise: float | None = 0.1

    @override
    def add_model_defaults(self, req: LLMRequest) -> LLMRequest:
        return self.model.add_model_defaults(req)

    @override
    def estimate_budget(self, req: LLMRequest) -> Budget:
        return self.model.estimate_budget(req)

    def retry_delays(self) -> Iterable[float]:
        import random

        acc = self.base_delay_seconds
        for _ in range(self.num_attempts):
            delay = acc
            if self.delay_noise is not None:
                delay += random.uniform(0, self.delay_noise)
            yield delay
            acc *= self.exponential_factor

    @override
    def _send_final_request(self, req: LLMRequest) -> LLMResponse:
        for i, retry_delay in enumerate([*self.retry_delays(), None]):
            try:
                ret = self.model.send_request(req, None)
                if i > 0:
                    ret.log_items.append(
                        LLMResponseLogItem(
                            "info", "successful_retry", {"delay": retry_delay}
                        )
                    )
                return ret
            except LLMBusyException as e:
                if retry_delay is None:
                    raise e
                else:
                    time.sleep(retry_delay)
        assert False

CachedModel dataclass

Bases: LLM

Wrap a model to use a given cache.

Note

The LLM.send_request method has a cache argument that can be used as a replacement for the CachedModel wrapper. In addition, all standard prompting policies use a global request cache (see PolicyEnv) when available. Thus, external users should rarely need to manually wrap models with CachedModel.

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

@dataclass
class CachedModel(LLM):
    """
    Wrap a model to use a given cache.

    !!! note
        The `LLM.send_request` method has a `cache` argument that can be
        used as a replacement for the `CachedModel` wrapper. In
        addition, all standard prompting policies use a global request
        cache (see `PolicyEnv`) when available. Thus, external users
        should rarely need to manually wrap models with `CachedModel`.
    """

    model: LLM
    cache: LLMCache

    def __post_init__(self):
        @cache(
            cache_spec=self.cache.spec,
            hash_arg=_CachedRequest.stable_repr,
        )
        def run_request(req: _CachedRequest) -> LLMResponse:
            base = req.request
            return self.model.send_request(base, None)

        self.run_request = run_request

    @override
    def _send_final_request(self, req: LLMRequest) -> LLMResponse:
        self.cache.num_seen[req] += 1
        num_seen = self.cache.num_seen[req]
        return self.run_request(_CachedRequest(req, num_seen))

    @override
    def estimate_budget(self, req: LLMRequest) -> Budget:
        return self.model.estimate_budget(req)

    @override
    def add_model_defaults(self, req: LLMRequest) -> LLMRequest:
        return self.model.add_model_defaults(req)

LLMCache dataclass

A cache for LLM requests.

More precisely, what are cached are (r, i) pairs where r is a request and i is the number of times the request has been answered since the model was instantiated. This way, caching works even when a policy samples multiple answers for the same request.

Multiple models can share the same cache.

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

@dataclass
class LLMCache:
    """
    A cache for LLM requests.

    More precisely, what are cached are `(r, i)` pairs where `r` is a
    request and `i` is the number of times the request has been answered
    since the model was instantiated. This way, caching works even when
    a policy samples multiple answers for the same request.

    Multiple models can share the same cache.
    """

    spec: CacheSpec
    num_seen: dict[LLMRequest, int]

    def __init__(self, spec: CacheSpec):
        self.spec = spec
        self.num_seen: dict[LLMRequest, int] = defaultdict(lambda: 0)

Token dataclass

A token produced by an LLM.

Attributes:

Name	Type	Description
token	str	String representation of the token.
bytes	Sequence[int] | None	Optional sequence of integers representing the token's byte encoding.

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

@dataclass
class Token:
    """
    A token produced by an LLM.

    Attributes:
        token: String representation of the token.
        bytes: Optional sequence of integers representing the token's
            byte encoding.
    """

    token: str
    bytes: Sequence[int] | None

TokenInfo dataclass

Logprob information for a single token.

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

@dataclass
class TokenInfo:
    """
    Logprob information for a single token.
    """

    token: Token
    logprob: float
    top_logprobs: Sequence[tuple[Token, float]] | None

ModelPricing dataclass

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

@dataclass
class ModelPricing:
    dollars_per_input_token: float
    dollars_per_cached_input_token: float
    dollars_per_output_token: float

LLMResponseLogItem dataclass

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

@dataclass
class LLMResponseLogItem:
    severity: Literal["info", "warning", "error"]
    message: str
    metadata: Any = None

LLMBusyException dataclass

Bases: Exception

This exception should be raised when an LLM call failed due to a timeout or a rate limit error that warrants a retry. In particular, it should not be raised for ill-formed requests (those assumptions should not be caught) or when the LLM gave a bad answer (in which case budget was consumed and should be counted, while errors are added into LLMResponse).

See WithRetry for adding retrial logic to LLMs.

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

@dataclass
class LLMBusyException(Exception):
    """
    This exception should be raised when an LLM call failed due to a
    timeout or a rate limit error that warrants a retry. In particular,
    it should not be raised for ill-formed requests (those assumptions
    should not be caught) or when the LLM gave a bad answer (in which
    case budget was consumed and should be counted, while errors are
    added into `LLMResponse`).

    See `WithRetry` for adding retrial logic to LLMs.
    """

    exn: Exception

    def __str__(self) -> str:
        return str(self.exn)

BudgetCategory

BudgetCategory: Literal[
    "num_requests",
    "num_completions",
    "input_tokens",
    "cached_input_tokens",
    "output_tokens",
    "price",
]

Standard metrics to measure LLM inference usage.

budget_entry

budget_entry(category: BudgetCategory, model_class: str | None = None) -> str

Return a string that can be used as a key in a budget dictionary.

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

def budget_entry(
    category: BudgetCategory, model_class: str | None = None
) -> str:
    """
    Return a string that can be used as a key in a budget dictionary.
    """
    res = category
    if model_class is not None:
        res = f"{res}{BUDGET_ENTRY_SEPARATOR}{model_class}"
    return res

NUM_REQUESTS module-attribute

NUM_REQUESTS = 'num_requests'

NUM_COMPLETIONS module-attribute

NUM_COMPLETIONS = 'num_completions'

NUM_INPUT_TOKENS module-attribute

NUM_INPUT_TOKENS = 'input_tokens'

NUM_CACHED_INPUT_TOKENS module-attribute

NUM_CACHED_INPUT_TOKENS = 'cached_input_tokens'

NUM_OUTPUT_TOKENS module-attribute

NUM_OUTPUT_TOKENS = 'output_tokens'

DOLLAR_PRICE module-attribute

DOLLAR_PRICE = 'price'