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 calls. 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 calls. 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()
        example_id = extra_args.get("example_id") if extra_args else None
        assert example_id is None or isinstance(example_id, int)
        assert not extra_args or "example" not in extra_args
        args_min: QueryTemplateArgs = {
            "query": self,
            "mode": mode,
            "available_modes": self.query_modes(),
            "params": params,
            "format": self._instantiated_parser_for(mode).formatting,
            "example_id": example_id,
            "example": example_id is not None,
        }
        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

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

    ### Hindsight Feedback

    @override
    def unparse(self, value: T) -> Answer | None:
        """
        Unparse an answer.

        By default, this method checks whether there is a single mode
        that uses structured output, and handles this case. This method
        can be overriden to handle more advanced cases.
        """

        modes = self.query_modes()
        if len(modes) != 1:
            return None
        mode = modes[0]
        parser = self._instantiated_parser_for(mode)
        if parser.settings.structured_output is None:
            return None
        output_type = parser.settings.structured_output.type
        if isinstance(output_type, ty.NoTypeInfo):
            return None
        structured = ty.pydantic_dump(output_type, value)
        answer = dp.Answer(mode, dp.Structured(structured))
        return answer

    ### 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

unparse

unparse(value: T) -> Answer | None

Unparse an answer.

By default, this method checks whether there is a single mode that uses structured output, and handles this case. This method can be overriden to handle more advanced cases.

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

@override
def unparse(self, value: T) -> Answer | None:
    """
    Unparse an answer.

    By default, this method checks whether there is a single mode
    that uses structured output, and handles this case. This method
    can be overriden to handle more advanced cases.
    """

    modes = self.query_modes()
    if len(modes) != 1:
        return None
    mode = modes[0]
    parser = self._instantiated_parser_for(mode)
    if parser.settings.structured_output is None:
        return None
    output_type = parser.settings.structured_output.type
    if isinstance(output_type, ty.NoTypeInfo):
        return None
    structured = ty.pydantic_dump(output_type, value)
    answer = dp.Answer(mode, dp.Structured(structured))
    return answer

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]

    @staticmethod
    def pure[G](value: G) -> "Response[G, Any]":
        """
        A response with an dummy value for the `answer` field.

        This is useful in particular for providing feedback for queries
        that have a `Response` answer type. In this case, only the final
        parsed answer matters.
        """
        answer = dp.Answer(None, dp.Structured(None))
        parsed = FinalAnswer(value)
        return Response(answer, parsed)

    def unwrap(self) -> F:
        """
        Extract a final parsed answer.

        Error if the response contains tool calls instead.
        """
        if isinstance(self.parsed, FinalAnswer):
            return self.parsed.final
        else:
            raise RuntimeError("Not a final answer.")

pure staticmethod

pure(value: G) -> Response[G, Any]

A response with an dummy value for the answer field.

This is useful in particular for providing feedback for queries that have a Response answer type. In this case, only the final parsed answer matters.

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

@staticmethod
def pure[G](value: G) -> "Response[G, Any]":
    """
    A response with an dummy value for the `answer` field.

    This is useful in particular for providing feedback for queries
    that have a `Response` answer type. In this case, only the final
    parsed answer matters.
    """
    answer = dp.Answer(None, dp.Structured(None))
    parsed = FinalAnswer(value)
    return Response(answer, parsed)

unwrap

unwrap() -> F

Extract a final parsed answer.

Error if the response contains tool calls instead.

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

def unwrap(self) -> F:
    """
    Extract a final parsed answer.

    Error if the response contains tool calls instead.
    """
    if isinstance(self.parsed, FinalAnswer):
        return self.parsed.final
    else:
        raise RuntimeError("Not a final answer.")

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. In a multi-message chat with few-shot examples, this variable can have different values across examples. Also, it has the same value for the system prompt and the final instance prompt.
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, as derived from mode (and whose value may therefore differ across examples).
example_id	int | None	If the message is part of an example, indicate the example number (examples are numbered starting from 1). Otherwise, indicate None. For example, this can be used to prefix each example with a # Example {i} header.
example	bool	Whether or not the message is part of an example (redundant with example_id, provided for readability). For example, when using a hyperparameter from params to provide specific answer instructions, one may want to include the specific instructions only in the final instance prompt and not in examples.

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. In a multi-message chat with
            few-shot examples, this variable can have different values
            across examples. Also, it has the same value for the system
            prompt and the final instance prompt.
        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, as derived from `mode` (and whose
            value may therefore differ across examples).
        example_id: If the message is part of an example, indicate the
            example number (examples are numbered starting from 1).
            Otherwise, indicate `None`. For example, this can be used to
            prefix each example with a `# Example {i}` header.
        example: Whether or not the message is part of an example
            (redundant with `example_id`, provided for readability). For
            example, when using a hyperparameter from `params` to
            provide specific answer instructions, one may want to
            include the specific instructions only in the final instance
            prompt and not in examples.
    """

    # 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
    example_id: int | None
    example: bool

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: ExampleSelector | None = None,
    mode: AnswerMode = None,
    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	ExampleSelector | None	Example selector (default: all_examples).	None
mode	AnswerMode	The answer mode to use for parsing the query answer.	None
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: ExampleSelector | None = None,
    mode: dp.AnswerMode = None,
    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: Example selector (default: `all_examples`).
        mode: The answer mode to use for parsing the query answer.
        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

    def parse(answer: dp.Answer) -> dp.Tracked[T] | dp.ParseError:
        parsed = query.parse_answer(answer)
        if no_wrap_parse_errors:
            parsed = _unwrap_parse_error(parsed)
        return parsed

    env.tracer.trace_query(query)

    if (overriden := env.overriden_answer(query.query)) is not None:
        overriden_parsed = parse(overriden)
        assert not isinstance(overriden_parsed, dp.ParseError), (
            f"Unexpected parse error: {overriden_parsed}"
        )
        yield dp.Solution(overriden_parsed)
        return

    if select_examples is None:
        select_examples = all_examples
    examples = select_examples(env, query.query)
    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(
            chat=prompt,
            num_completions=num_completions,
            options=options,
            tools=tuple(tools),
            structured_output=structured_output,
        )
        resp = yield from _send_request(
            env, model=model, request=req, query=query
        )
        if isinstance(resp, SpendingDeclined):
            return
        if not resp.outputs:
            env.warn("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 = parse(answer)
            env.tracer.trace_answer(query.ref, answer)
            if isinstance(element, dp.ParseError):
                env.info("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: ExampleSelector | None = None,
    mode: AnswerMode = None,
    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	ExampleSelector | None	Example selector. By default, all_examples is used.	None
mode	AnswerMode	The answer mode to use for parsing the query answer.	None
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: ExampleSelector | None = None,
    mode: dp.AnswerMode = None,
    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. By default, `all_examples` is
            used.
        mode: The answer mode to use for parsing the query answer.
        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)
    if select_examples is None:
        select_examples = all_examples
    examples = select_examples(env, query.query)
    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(
        chat=prompt,
        num_completions=1,
        options=options,
    )
    resp = yield from _send_request(env, model=model, request=req, query=query)
    if isinstance(resp, SpendingDeclined):
        return
    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]]

ExampleSelector dataclass

Wrapper around a function for selecting examples.

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

@dataclass(frozen=True)
class ExampleSelector:
    """
    Wrapper around a function for selecting examples.
    """

    _fn: _ExampleSelectorFn

    def __call__(
        self, env: PolicyEnv, query: dp.AbstractQuery[Any]
    ) -> Sequence[Example]:
        return [e.example for e in self._fn(env, query)]

    def filter(self, f: ExampleFilter) -> "ExampleSelector":
        """
        Return a new example selector that applies a filter to the
        examples selected by this selector.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
        ) -> Sequence[SelectedExample]:
            examples = self._fn(env, query)
            return f(env, query, examples)

        return ExampleSelector(select)

    def concat(self, other: "ExampleSelector") -> "ExampleSelector":
        """
        Return a new example selector that concatenates the examples
        selected by this selector and another selector.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
        ) -> Sequence[SelectedExample]:
            examples1 = self._fn(env, query)
            examples2 = other._fn(env, query)
            return [*examples1, *examples2]

        return ExampleSelector(select)

    def deduplicate(self) -> "ExampleSelector":
        """
        Return a new example selector that removes duplicate examples
        while preserving order.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
            examples: Sequence[SelectedExample],
        ) -> Sequence[SelectedExample]:
            seen: set[int] = set()
            deduped: list[SelectedExample] = []
            for ex in examples:
                if ex.index not in seen:
                    seen.add(ex.index)
                    deduped.append(ex)
            return deduped

        return self.filter(select)

    def reverse(self) -> "ExampleSelector":
        """
        Return a new example selector that reverses the order of the
        selected examples.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
            examples: Sequence[SelectedExample],
        ) -> Sequence[SelectedExample]:
            return list(reversed(examples))

        return self.filter(select)

    def __add__(self, other: "ExampleSelector") -> "ExampleSelector":
        """
        Concatenate two example selectors and deduplicate the result.
        """
        return self.concat(other).deduplicate()

    def random(self, num_examples: int) -> "ExampleSelector":
        """
        Return a new example selector that randomly selects a given
        number of examples.
        """
        return self.filter(take_random(num_examples))

    def with_tags(self, tags: Sequence[str]) -> "ExampleSelector":
        """
        Return a new example selector that only includes examples
        having all the given tags.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
            examples: Sequence[SelectedExample],
        ) -> Sequence[SelectedExample]:
            return [
                ex
                for ex in examples
                if all(tag in ex.example.tags for tag in tags)
            ]

        return self.filter(select)

    def exclude_identical_queries(self) -> "ExampleSelector":
        """
        Return a new example selector that excludes examples whose
        query is identical to the input query.
        """

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
            examples: Sequence[SelectedExample],
        ) -> Sequence[SelectedExample]:
            serialized = dp.SerializedQuery.make(query)
            return [
                ex
                for ex in examples
                if not dp.SerializedQuery.make(ex.example.query) == serialized
            ]

        return self.filter(select)

    def cached(self) -> "ExampleSelector":
        """
        Return a new example selector that caches its output the first
        time it is called and then always returns the cached result
        regardless of its inputs.

        This is useful in particular when using `interact`, so that each
        query that is part of a same conversation is mapped to the same
        set of examples.
        """

        cached: Sequence[SelectedExample] | None = None

        def select(
            env: PolicyEnv,
            query: dp.AbstractQuery[Any],
        ) -> Sequence[SelectedExample]:
            nonlocal cached
            if cached is None:
                cached = self._fn(env, query)
            return cached

        return ExampleSelector(select)

filter

filter(f: ExampleFilter) -> ExampleSelector

Return a new example selector that applies a filter to the examples selected by this selector.

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

def filter(self, f: ExampleFilter) -> "ExampleSelector":
    """
    Return a new example selector that applies a filter to the
    examples selected by this selector.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
    ) -> Sequence[SelectedExample]:
        examples = self._fn(env, query)
        return f(env, query, examples)

    return ExampleSelector(select)

concat

concat(other: ExampleSelector) -> ExampleSelector

Return a new example selector that concatenates the examples selected by this selector and another selector.

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

def concat(self, other: "ExampleSelector") -> "ExampleSelector":
    """
    Return a new example selector that concatenates the examples
    selected by this selector and another selector.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
    ) -> Sequence[SelectedExample]:
        examples1 = self._fn(env, query)
        examples2 = other._fn(env, query)
        return [*examples1, *examples2]

    return ExampleSelector(select)

deduplicate

deduplicate() -> ExampleSelector

Return a new example selector that removes duplicate examples while preserving order.

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

def deduplicate(self) -> "ExampleSelector":
    """
    Return a new example selector that removes duplicate examples
    while preserving order.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
        examples: Sequence[SelectedExample],
    ) -> Sequence[SelectedExample]:
        seen: set[int] = set()
        deduped: list[SelectedExample] = []
        for ex in examples:
            if ex.index not in seen:
                seen.add(ex.index)
                deduped.append(ex)
        return deduped

    return self.filter(select)

reverse

reverse() -> ExampleSelector

Return a new example selector that reverses the order of the selected examples.

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

def reverse(self) -> "ExampleSelector":
    """
    Return a new example selector that reverses the order of the
    selected examples.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
        examples: Sequence[SelectedExample],
    ) -> Sequence[SelectedExample]:
        return list(reversed(examples))

    return self.filter(select)

__add__

__add__(other: ExampleSelector) -> ExampleSelector

Concatenate two example selectors and deduplicate the result.

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

def __add__(self, other: "ExampleSelector") -> "ExampleSelector":
    """
    Concatenate two example selectors and deduplicate the result.
    """
    return self.concat(other).deduplicate()

random

random(num_examples: int) -> ExampleSelector

Return a new example selector that randomly selects a given number of examples.

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

def random(self, num_examples: int) -> "ExampleSelector":
    """
    Return a new example selector that randomly selects a given
    number of examples.
    """
    return self.filter(take_random(num_examples))

with_tags

with_tags(tags: Sequence[str]) -> ExampleSelector

Return a new example selector that only includes examples having all the given tags.

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

def with_tags(self, tags: Sequence[str]) -> "ExampleSelector":
    """
    Return a new example selector that only includes examples
    having all the given tags.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
        examples: Sequence[SelectedExample],
    ) -> Sequence[SelectedExample]:
        return [
            ex
            for ex in examples
            if all(tag in ex.example.tags for tag in tags)
        ]

    return self.filter(select)

exclude_identical_queries

exclude_identical_queries() -> ExampleSelector

Return a new example selector that excludes examples whose query is identical to the input query.

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

def exclude_identical_queries(self) -> "ExampleSelector":
    """
    Return a new example selector that excludes examples whose
    query is identical to the input query.
    """

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
        examples: Sequence[SelectedExample],
    ) -> Sequence[SelectedExample]:
        serialized = dp.SerializedQuery.make(query)
        return [
            ex
            for ex in examples
            if not dp.SerializedQuery.make(ex.example.query) == serialized
        ]

    return self.filter(select)

cached

cached() -> ExampleSelector

Return a new example selector that caches its output the first time it is called and then always returns the cached result regardless of its inputs.

This is useful in particular when using interact, so that each query that is part of a same conversation is mapped to the same set of examples.

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

def cached(self) -> "ExampleSelector":
    """
    Return a new example selector that caches its output the first
    time it is called and then always returns the cached result
    regardless of its inputs.

    This is useful in particular when using `interact`, so that each
    query that is part of a same conversation is mapped to the same
    set of examples.
    """

    cached: Sequence[SelectedExample] | None = None

    def select(
        env: PolicyEnv,
        query: dp.AbstractQuery[Any],
    ) -> Sequence[SelectedExample]:
        nonlocal cached
        if cached is None:
            cached = self._fn(env, query)
        return cached

    return ExampleSelector(select)

all_examples

all_examples(env: PolicyEnv, query: AbstractQuery[Any]) -> Sequence[SelectedExample]

Select all examples relevant to a query.

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

@ExampleSelector
def all_examples(
    env: PolicyEnv, query: dp.AbstractQuery[Any]
) -> Sequence[SelectedExample]:
    """
    Select all examples relevant to a query.
    """
    examples = env.examples.examples_for(query.query_name())
    return [
        SelectedExample(example=e, index=i, similarity=None)
        for i, e in enumerate(examples)
    ]

closest_examples

closest_examples(*, k: int, model_name: str) -> ExampleSelector

Obtain the k closest examples to the given query, based on embedding similarity. Return a list of (example, similarity) tuples, sorted by decreasing similarity.

Parameters:

Name	Type	Description	Default
k	int	Number of examples to select.	required
model_name	str	Name of the embedding model to use.	required

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

def closest_examples(
    *,
    k: int,
    model_name: str,
) -> ExampleSelector:
    """
    Obtain the `k` closest examples to the given query, based on
    embedding similarity. Return a list of `(example, similarity)`
    tuples, sorted by decreasing similarity.

    Arguments:
        k: Number of examples to select.
        model_name: Name of the embedding model to use.
    """

    def select(
        env: PolicyEnv, query: dp.AbstractQuery[Any]
    ) -> Sequence[SelectedExample]:
        qname = query.query_name()
        lid = env.info(
            "closest_examples_request",
            {"query": qname},
        )
        embeddings = env.examples.fetch_query_embeddings(qname, model_name)
        if embeddings is None:
            # There are no examples in the database.
            return []
        model = em.standard_openai_embedding_model(model_name)
        query_embedding = model.embed(
            [env.examples.query_embedding_text(query)],
            cache=env.embeddings_cache,
        )[0].embedding
        # Compute cosine similarities.
        sims = np.dot(embeddings, query_embedding) / (
            np.linalg.norm(embeddings, axis=1)
            * np.linalg.norm(query_embedding)
        )
        # Get top-k indices.
        top_k_indices = cast(list[int], np.argsort(-sims)[:k].tolist())
        examples = env.examples.examples_for(qname)
        env.info(
            "closest_examples_response",
            related=[lid],
            metadata={
                "query": qname,
                "args": query.serialize_args(),
                "selected": [
                    {
                        "similarity": float(sims[k]),
                        "args": examples[k].query.serialize_args(),
                    }
                    for k in top_k_indices
                ],
            },
        )
        return [
            SelectedExample(
                example=examples[i], index=i, similarity=float(sims[i])
            )
            for i in top_k_indices
        ]

    return ExampleSelector(select)

maximum_marginally_relevant

maximum_marginally_relevant(
    *, k: int, lambda_param: float, model_name: str, always_compute_mmr: bool = False
) -> ExampleSelector

Obtain k examples that are maximally relevant to the query while being diverse among themselves, using the Maximal Marginally Relevant (MMR) algorithm.

Parameters:

Name	Type	Description	Default
k	int	Number of examples to select.	required
lambda_param	float	Trade-off parameter between relevance and diversity, in [0, 1]. Higher values favor relevance.	required
model_name	str	Name of the embedding model to use. always_compute_mmr: If True, the MMR algorithm is run even when all examples need to be returned. This is useful for debugging purposes or to ensure similarity scores are attached to the output.	required

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

def maximum_marginally_relevant(
    *,
    k: int,
    lambda_param: float,
    model_name: str,
    always_compute_mmr: bool = False,
) -> ExampleSelector:
    """
    Obtain `k` examples that are maximally relevant to the query while
    being diverse among themselves, using the Maximal Marginally
    Relevant (MMR) algorithm.

    Arguments:
        k: Number of examples to select.
        lambda_param: Trade-off parameter between relevance and
            diversity, in [0, 1]. Higher values favor relevance.
        model_name: Name of the embedding model to use.
            always_compute_mmr: If `True`, the MMR algorithm is run even
            when all examples need to be returned. This is useful for
            debugging purposes or to ensure similarity scores are
            attached to the output.
    """

    def select(
        env: PolicyEnv, query: dp.AbstractQuery[Any]
    ) -> Sequence[SelectedExample]:
        if k == 0:
            return []

        qname = query.query_name()
        examples = env.examples.examples_for(qname)
        num_examples = len(examples)
        if k >= num_examples and not always_compute_mmr:
            # Return all examples if we need more than available
            return [
                SelectedExample(example=e, index=i, similarity=None)
                for i, e in enumerate(examples)
            ]

        lid = env.info(
            "mmr_request",
            {"query": qname, "k": k, "lambda": lambda_param},
        )
        embeddings = env.examples.fetch_query_embeddings(qname, model_name)
        similarity_matrix = env.examples.fetch_example_similarity_matrix(
            qname, model_name
        )
        if embeddings is None or similarity_matrix is None:
            return []

        model = em.standard_openai_embedding_model(model_name)
        query_embedding = model.embed(
            [env.examples.query_embedding_text(query)],
            cache=env.embeddings_cache,
        )[0].embedding

        # Compute cosine similarities to the query for all examples
        query_sims = np.dot(embeddings, query_embedding) / (
            np.linalg.norm(embeddings, axis=1)
            * np.linalg.norm(query_embedding)
        )

        # MMR algorithm
        selected_indices: list[int] = []
        remaining_indices = set(range(num_examples))
        max_similarities: dict[
            int, float
        ] = {}  # Store max similarity for each selected index
        mmr_scores: dict[
            int, float
        ] = {}  # Store MMR score for each selected index

        # Select the first example with highest query similarity
        first_idx = int(np.argmax(query_sims))
        selected_indices.append(first_idx)
        remaining_indices.remove(first_idx)
        max_similarities[first_idx] = 0.0
        mmr_scores[first_idx] = query_sims[first_idx]

        # Iteratively select remaining examples
        for _ in range(k - 1):
            if not remaining_indices:
                break
            best_score = float("-inf")
            best_idx = -1
            best_max_similarity = 0.0
            for idx in remaining_indices:
                relevance = query_sims[idx]
                assert selected_indices
                max_similarity = np.max(
                    similarity_matrix[idx, selected_indices]
                )
                mmr_score = (
                    lambda_param * relevance
                    - (1 - lambda_param) * max_similarity
                )
                if mmr_score > best_score:
                    best_score = mmr_score
                    best_idx = idx
                    best_max_similarity = max_similarity
            selected_indices.append(best_idx)
            remaining_indices.remove(best_idx)
            max_similarities[best_idx] = float(best_max_similarity)
            mmr_scores[best_idx] = best_score

        env.info(
            "mmr_response",
            related=[lid],
            metadata={
                "query": qname,
                "args": query.serialize_args(),
                "selected": [
                    {
                        "score": float(mmr_scores[idx]),
                        "query_similarity": float(query_sims[idx]),
                        "max_example_similarity": float(max_similarities[idx]),
                        "args": examples[idx].query.serialize_args(),
                    }
                    for idx in selected_indices
                ],
            },
        )
        return [
            SelectedExample(
                example=examples[i], index=i, similarity=float(query_sims[i])
            )
            for i in selected_indices
        ]

    return ExampleSelector(select)

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.
        """
        full_req = self.add_model_defaults(req)
        return fetch_or_answer(cache, full_req, self._send_final_request)

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.
    """
    full_req = self.add_model_defaults(req)
    return fetch_or_answer(cache, full_req, self._send_final_request)

LLMRequest dataclass

An LLM chat completion request.

Attributes:

Name	Type	Description
chat	Chat	The chat history.
options	RequestOptions	Request options.
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.
tools	tuple[Schema, ...]	Available tools.
structured_output	Schema | None	Provide a schema to enable structured output, or None for disabling it.

Note

This class is hashable, as needed by LLMCache. For soundness, it is assumed that RequestOptions dictionaries are immutable.

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

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

    Attributes:
        chat: The chat history.
        options: Request options.
        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.
        tools: Available tools.
        structured_output: Provide a schema to enable structured output,
            or `None` for disabling it.

    !!! note
        This class is hashable, as needed by `LLMCache`. For soundness,
        it is assumed that `RequestOptions` dictionaries are immutable.
    """

    chat: Chat
    options: RequestOptions
    num_completions: int = 1
    tools: tuple[Schema, ...] = ()
    structured_output: Schema | None = None

    def _hashable_repr(self) -> Any:
        import json

        return (
            self.chat,
            self.num_completions,
            json.dumps(self.options, sort_keys=True),
            self.tools,
            self.structured_output,
        )

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, LLMRequest):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

LLMResponse dataclass

Response to an LLM request.

Attributes:

Name	Type	Description
outputs	Sequence[LLMOutput]	Generated completions.
budget	Budget | None	Budget consumed by the request.
log_items	Sequence[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 | None = None
    log_items: Sequence[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 = tuple[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	ReasoningEffort	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.

Warning

Dictionaries of this type should be treated as immutable, since they are used as part of the hash of LLMRequest objects.

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.

    !!! warning
        Dictionaries of this type should be treated as immutable, since
        they are used as part of the hash of `LLMRequest` objects.
    """

    model: str
    reasoning_effort: ReasoningEffort
    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(),
        )

    def _hashable_repr(self) -> str:
        # See comment in ToolCall._hashable_repr
        import json

        return json.dumps(self.__dict__, sort_keys=True)

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Schema):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

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 = "stop"
    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:
                    msg = LLMResponseLogItem(
                        "info", "successful_retry", {"delay": retry_delay}
                    )
                    ret.log_items = (*ret.log_items, msg)
                return ret
            except LLMBusyException as e:
                if retry_delay is None:
                    raise e
                else:
                    time.sleep(retry_delay)
        assert False

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.

LLMCache objects can be created using the load_request_cache context manager.

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.

    `LLMCache` objects can be created using the `load_request_cache`
    context manager.
    """

    cache: Cache[CachedRequest, LLMResponse]
    num_seen: dict[LLMRequest, int]

    def __init__(self, cache: Cache[CachedRequest, LLMResponse]):
        self.cache = cache
        self.num_seen: dict[LLMRequest, int] = defaultdict(lambda: 0)

load_request_cache

load_request_cache(file: Path, *, mode: CacheMode)

Context manager that loads an LLM request cache from a YAML file.

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

@contextmanager
def load_request_cache(file: Path, *, mode: CacheMode):
    """
    Context manager that loads an LLM request cache from a YAML file.
    """
    with load_cache(
        file, mode=mode, input_type=CachedRequest, output_type=LLMResponse
    ) as cache:
        yield LLMCache(cache)

CacheMode

CacheMode = Literal['read_write', 'off', 'create', 'replay']

Caching mode:

• off: the cache is disabled.
• read_write: values can be read and written to the cache (no extra check is made).
• create: the cache is used in write-only mode, and an exception is raised if a cached value already exists.
• replay: all requests must hit the cache or an exception is raised.

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:
    level: LogLevel
    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'

Standard Models

standard_model

standard_model(
    model: StandardModelName | str,
    options: RequestOptions | None = None,
    *,
    pricing: ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
) -> OpenAICompatibleModel

Obtain a standard model from OpenAI, Mistral, DeepSeek or Gemini.

Make sure that the following environment variables are set:

• OPENAI_API_KEY for OpenAI models
• MISTRAL_API_KEY for Mistral models
• DEEPSEEK_API_KEY for DeepSeek models
• GEMINI_API_KEY for Gemini models

Parameters:

Name	Type	Description	Default
model	StandardModelName | str	The name of the model to use. The model provider is automatically inferred from this name.	required
options	RequestOptions | None	Additional options for the model, such as reasoning effort or default temperature. The model option must not be overriden.	None
pricing	ModelPricing | None | Literal['auto']	Pricing model to use. If "auto" is provided (default), it is inferred from the model's name (or ValueError is raised). If None is provided, no pricing information is used and so the associated budget metrics won't be computed.	'auto'
model_class	str | None	An optional identifier for the model class (e.g., "reasoning_large"). When provided, class-specific budget metrics are reported, so that resource consumption can be tracked separately for different classes of models (e.g., tracking "num_requests__reasoning_large" separately from "num_requests__chat_small").	None

Raises:

Type	Description
ValueError	The provider or pricing model could not be inferred.

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

def standard_model(
    model: StandardModelName | str,
    options: md.RequestOptions | None = None,
    *,
    pricing: md.ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
) -> OpenAICompatibleModel:
    """
    Obtain a standard model from OpenAI, Mistral, DeepSeek or Gemini.

    Make sure that the following environment variables are set:

    - `OPENAI_API_KEY` for OpenAI models
    - `MISTRAL_API_KEY` for Mistral models
    - `DEEPSEEK_API_KEY` for DeepSeek models
    - `GEMINI_API_KEY` for Gemini models

    Parameters:
        model: The name of the model to use. The model provider is
            automatically inferred from this name.
        options: Additional options for the model, such as reasoning
            effort or default temperature. The `model` option must not
            be overriden.
        pricing: Pricing model to use. If `"auto"` is provided
            (default), it is inferred from the model's name (or
            `ValueError` is raised). If `None` is provided, no pricing
            information is used and so the associated budget metrics
            won't be computed.
        model_class: An optional identifier for the model class (e.g.,
            "reasoning_large"). When provided, class-specific budget
            metrics are reported, so that resource consumption can be
            tracked separately for different classes of models (e.g.,
            tracking "num_requests__reasoning_large" separately from
            "num_requests__chat_small").

    Raises:
        ValueError: The provider or pricing model could not be inferred.
    """

    openai_models = _values(OpenAIModelName)
    mistral_models = _values(MistralModelName)
    deepseek_models = _values(DeepSeekModelName)
    gemini_models = _values(GeminiModelName)

    prefix = _longest_standard_model_prefix_or_self(model)

    if prefix in openai_models:
        make_model = openai_model
    elif prefix in mistral_models:
        make_model = mistral_model
    elif prefix in deepseek_models:
        make_model = deepseek_model
    elif prefix in gemini_models:
        make_model = gemini_model
    else:
        raise ValueError(
            f"Failed to infer provider for model: {model}.\n"
            + "Use a more specific function such as `openai_model`."
        )
    return make_model(
        model, options=options, pricing=pricing, model_class=model_class
    )

openai_model

openai_model(
    model: OpenAIModelName | str,
    options: RequestOptions | None = None,
    *,
    pricing: ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
)

Obtain a standard model from OpenAI.

See standard_model for details.

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

def openai_model(
    model: OpenAIModelName | str,
    options: md.RequestOptions | None = None,
    *,
    pricing: md.ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
):
    """
    Obtain a standard model from OpenAI.

    See `standard_model` for details.
    """
    return _openai_compatible_model(
        model,
        options=options,
        pricing=pricing,
        model_class=model_class,
        base_url="https://api.openai.com/v1",
        api_key_env_var="OPENAI_API_KEY",
    )

mistral_model

mistral_model(
    model: MistralModelName | str,
    options: RequestOptions | None = None,
    *,
    pricing: ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
)

Obtain a standard model from Mistral.

See standard_model for details.

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

def mistral_model(
    model: MistralModelName | str,
    options: md.RequestOptions | None = None,
    *,
    pricing: md.ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
):
    """
    Obtain a standard model from Mistral.

    See `standard_model` for details.
    """
    return _openai_compatible_model(
        model,
        options=options,
        pricing=pricing,
        model_class=model_class,
        base_url="https://api.mistral.ai/v1",
        api_key_env_var="MISTRAL_API_KEY",
    )

deepseek_model

deepseek_model(
    model: DeepSeekModelName | str,
    options: RequestOptions | None = None,
    *,
    pricing: ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
)

Obtain a standard model from DeepSeek.

See standard_model for details.

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

def deepseek_model(
    model: DeepSeekModelName | str,
    options: md.RequestOptions | None = None,
    *,
    pricing: md.ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
):
    """
    Obtain a standard model from DeepSeek.

    See `standard_model` for details.
    """
    return _openai_compatible_model(
        model,
        options=options,
        pricing=pricing,
        model_class=model_class,
        base_url="https://api.deepseek.com",
        api_key_env_var="DEEPSEEK_API_KEY",
    )

gemini_model

gemini_model(
    model: GeminiModelName | str,
    options: RequestOptions | None = None,
    *,
    pricing: ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
)

Obtain a standard model from Gemini.

See standard_model for details.

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

def gemini_model(
    model: GeminiModelName | str,
    options: md.RequestOptions | None = None,
    *,
    pricing: md.ModelPricing | None | Literal["auto"] = "auto",
    model_class: str | None = None,
):
    """
    Obtain a standard model from Gemini.

    See `standard_model` for details.
    """
    return _openai_compatible_model(
        model,
        options=options,
        pricing=pricing,
        model_class=model_class,
        base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
        api_key_env_var="GEMINI_API_KEY",
    )