Policy Environments

Main Class

PolicyEnv

The global environment accessible to policies.

It can be used for:

Fetching prompts, data, and examples.
Caching LLM requests.
Tracing nodes, query, answers, and logging information.

Attributes:

Name	Type	Description
`requests_cache`	`LLMCache \| None`	The requests cache, or `None` if caching is disabled. The exact type of the request cache is not statically determined and depends on the value of constructor argument `make_cache`.
`templates`		The prompt templates manager.
`tracer`		The tracer, which can also be used for logging.
`examples`		The example database.

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

class PolicyEnv:
    """
    The global environment accessible to policies.

    It can be used for:

    - Fetching prompts, data, and examples.
    - Caching LLM requests.
    - Tracing nodes, query, answers, and logging information.

    Attributes:
        requests_cache: The requests cache, or `None` if caching is
            disabled. The exact type of the request cache is not
            statically determined and depends on the value of
            constructor argument `make_cache`.
        templates: The prompt templates manager.
        tracer: The tracer, which can also be used for logging.
        examples: The example database.
    """

    def __init__(
        self,
        *,
        prompt_dirs: Sequence[Path] = (),
        demonstration_files: Sequence[Path] = (),
        data_dirs: Sequence[Path] = (),
        cache: CacheSpec | None = None,
        do_not_match_identical_queries: bool = False,
    ):
        """
        Args:
            prompt_dirs: A sequence of directories where Jinja prompt
                templates can be found.
            demonstration_files: A sequence of paths to demonstration
                files (with or without extension `.demo.yaml`), to
                create an example database from.
            data_dirs: A sequence of directories where data files can be
                found.
            cache: A cache specification, or `None` to disable caching.
                When caching is enabled, the `requests_cache` attribute
                can be accessed by policies to properly wrap LLM models.
            do_not_match_identical_queries: See `ExampleDatabase`.
        """
        self.templates = TemplatesManager(prompt_dirs, data_dirs)
        self.examples = ExampleDatabase(do_not_match_identical_queries)
        self.tracer = Tracer()
        self.requests_cache: md.LLMCache | None = None
        if cache is not None:
            self.requests_cache = md.LLMCache(cache)
        for path in demonstration_files:
            if not path.suffix.endswith(DEMO_FILE_EXT):
                path = path.with_suffix(DEMO_FILE_EXT)
            try:
                with path.open() as f:
                    content = yaml.safe_load(f)
                    demos = pydantic_load(list[Demo], content)
                    for demo in demos:
                        self.examples.add_demonstration(demo)
            except Exception as e:
                raise InvalidDemoFile(path, e)

init

__init__(
    *,
    prompt_dirs: Sequence[Path] = (),
    demonstration_files: Sequence[Path] = (),
    data_dirs: Sequence[Path] = (),
    cache: CacheSpec | None = None,
    do_not_match_identical_queries: bool = False,
)

Parameters:

Name	Type	Description	Default
`prompt_dirs`	`Sequence[Path]`	A sequence of directories where Jinja prompt templates can be found.	`()`
`demonstration_files`	`Sequence[Path]`	A sequence of paths to demonstration files (with or without extension `.demo.yaml`), to create an example database from.	`()`
`data_dirs`	`Sequence[Path]`	A sequence of directories where data files can be found.	`()`
`cache`	`CacheSpec \| None`	A cache specification, or `None` to disable caching. When caching is enabled, the `requests_cache` attribute can be accessed by policies to properly wrap LLM models.	`None`
`do_not_match_identical_queries`	`bool`	See `ExampleDatabase`.	`False`

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

def __init__(
    self,
    *,
    prompt_dirs: Sequence[Path] = (),
    demonstration_files: Sequence[Path] = (),
    data_dirs: Sequence[Path] = (),
    cache: CacheSpec | None = None,
    do_not_match_identical_queries: bool = False,
):
    """
    Args:
        prompt_dirs: A sequence of directories where Jinja prompt
            templates can be found.
        demonstration_files: A sequence of paths to demonstration
            files (with or without extension `.demo.yaml`), to
            create an example database from.
        data_dirs: A sequence of directories where data files can be
            found.
        cache: A cache specification, or `None` to disable caching.
            When caching is enabled, the `requests_cache` attribute
            can be accessed by policies to properly wrap LLM models.
        do_not_match_identical_queries: See `ExampleDatabase`.
    """
    self.templates = TemplatesManager(prompt_dirs, data_dirs)
    self.examples = ExampleDatabase(do_not_match_identical_queries)
    self.tracer = Tracer()
    self.requests_cache: md.LLMCache | None = None
    if cache is not None:
        self.requests_cache = md.LLMCache(cache)
    for path in demonstration_files:
        if not path.suffix.endswith(DEMO_FILE_EXT):
            path = path.with_suffix(DEMO_FILE_EXT)
        try:
            with path.open() as f:
                content = yaml.safe_load(f)
                demos = pydantic_load(list[Demo], content)
                for demo in demos:
                    self.examples.add_demonstration(demo)
        except Exception as e:
            raise InvalidDemoFile(path, e)

InvalidDemoFile `dataclass`

Bases: Exception

Exception raised when a demonstration file could not be parsed.

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

@dataclass
class InvalidDemoFile(Exception):
    """
    Exception raised when a demonstration file could not be parsed.
    """

    file: Path
    exn: Exception

Example Database

Example `dataclass`

An example, usable for few-shot prompting.

Attributes:

Name	Type	Description
`args`	`QuerySerializedArgs`	The serialized query arguments.
`answer`	`Answer`	The answer to the query.
`tags`	`Sequence[str]`	A sequence of tags associated with the example, which policies can use to select appropriate examples.

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

@dataclass
class Example:
    """
    An example, usable for few-shot prompting.

    Attributes:
        args: The serialized query arguments.
        answer: The answer to the query.
        tags: A sequence of tags associated with the example, which
            policies can use to select appropriate examples.
    """

    args: QuerySerializedArgs
    answer: Answer
    tags: Sequence[str]

QuerySerializedArgs

QuerySerializedArgs: dict[str, Any]

Serialized query arguments, as a dictionary mapping attributed to JSON values (assemblies of integers, strings, dictionnaries, list, tuples...).

ExampleDatabase `dataclass`

A simple example database.

Attributes:

Name	Type	Description
`do_not_match_identical_queries`	`bool`	If set to `True`, the `examples` method won't return examples that match identical queries (i.e., with the exact same arguments). This is useful in the context of writing demonstrations, where one may want to see how an LLM would answer a query, even when a ground-truth answer is provided already.

TODO: add provenance info for better error messages.

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

@dataclass
class ExampleDatabase:
    """
    A simple example database.

    Attributes:
        do_not_match_identical_queries: If set to `True`, the `examples`
            method won't return examples that match identical queries
            (i.e., with the exact same arguments). This is useful in the
            context of writing demonstrations, where one may want to see
            how an LLM would answer a query, even when a ground-truth
            answer is provided already.

    TODO: add provenance info for better error messages.
    """

    do_not_match_identical_queries: bool = False

    # Maps each query name to a list of
    _examples: dict[_QueryName, list[Example]] = field(
        default_factory=lambda: defaultdict(list)
    )

    def add_query_demonstration(self, demo: QueryDemo):
        """
        Add all examples from a standalone query demonstration to the
        database.
        """
        if not demo.answers:
            return
        if (ex := demo.answers[0].example) is not None and not ex:
            # If the user explicitly asked not to include the example.
            # TODO: What if the user asked to include several answers?
            # Right now, we only allow the first one to be added.
            return
        demo_answer = demo.answers[0]
        answer = translate_answer(demo_answer)
        example = Example(demo.args, answer, demo_answer.tags)
        self._examples[demo.query].append(example)

    def add_demonstration(self, demo: Demo):
        """
        Add all exmples from a demonstration to the database.
        """
        if isinstance(demo, QueryDemo):
            self.add_query_demonstration(demo)
        else:
            assert isinstance(demo, StrategyDemo)
            for q in demo.queries:
                self.add_query_demonstration(q)

    def examples(
        self, query_name: str, query_args: QuerySerializedArgs
    ) -> Iterable[Example]:
        """
        Obtain all potential examples that can be used for few-shot
        prompting with a given query.
        """
        for ex in self._examples[query_name]:
            if self.do_not_match_identical_queries:
                if _equal_query_args(ex.args, query_args):
                    continue
            yield ex

add_query_demonstration

add_query_demonstration(demo: QueryDemo)

Add all examples from a standalone query demonstration to the database.

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

def add_query_demonstration(self, demo: QueryDemo):
    """
    Add all examples from a standalone query demonstration to the
    database.
    """
    if not demo.answers:
        return
    if (ex := demo.answers[0].example) is not None and not ex:
        # If the user explicitly asked not to include the example.
        # TODO: What if the user asked to include several answers?
        # Right now, we only allow the first one to be added.
        return
    demo_answer = demo.answers[0]
    answer = translate_answer(demo_answer)
    example = Example(demo.args, answer, demo_answer.tags)
    self._examples[demo.query].append(example)

add_demonstration

add_demonstration(demo: Demo)

Add all exmples from a demonstration to the database.

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

def add_demonstration(self, demo: Demo):
    """
    Add all exmples from a demonstration to the database.
    """
    if isinstance(demo, QueryDemo):
        self.add_query_demonstration(demo)
    else:
        assert isinstance(demo, StrategyDemo)
        for q in demo.queries:
            self.add_query_demonstration(q)

examples

examples(query_name: str, query_args: QuerySerializedArgs) -> Iterable[Example]

Obtain all potential examples that can be used for few-shot prompting with a given query.

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

def examples(
    self, query_name: str, query_args: QuerySerializedArgs
) -> Iterable[Example]:
    """
    Obtain all potential examples that can be used for few-shot
    prompting with a given query.
    """
    for ex in self._examples[query_name]:
        if self.do_not_match_identical_queries:
            if _equal_query_args(ex.args, query_args):
                continue
        yield ex

Tracer

TemplatesManager

Bases: AbstractTemplatesManager

A class for managing Jinja prompt templates.

Templates are configured with the trim_blocks and lstrip_blocks options set to True (no newlines are inserted after blocks and indentation can be used within blocks without affecting the output). The keep_trailing_newline option is set to False so trailing new lines at the end of template files are ignored.

Templates are first searched in the provided prompt folders and then in the standard library (delphyne.stdlib.templates). For example, to show standard formatting instructions, you can include the following in your instance prompts:

{% include 'stdlib/format.jinja' %}

All templates automatically have access to the following global objects:

A yaml filter for converting an object into a YAML string.
A json filter for converting an object into a JSON string.
A fail function that takes an error message as an argument and raises an exception on Python side.

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

class TemplatesManager(qu.AbstractTemplatesManager):
    """
    A class for managing Jinja prompt templates.

    Templates are configured with the `trim_blocks` and `lstrip_blocks`
    options set to `True` (no newlines are inserted after blocks and
    indentation can be used within blocks without affecting the output).
    The `keep_trailing_newline` option is set to `False` so trailing new
    lines at the end of template files are ignored.

    Templates are first searched in the provided prompt folders and then
    in the standard library (`delphyne.stdlib.templates`). For example,
    to show standard formatting instructions, you can include the
    following in your instance prompts:

    ```jinja
    {% include 'stdlib/format.jinja' %}
    ```

    All templates automatically have access to the following global
    objects:

    - A `yaml` filter for converting an object into a YAML string.
    - A `json` filter for converting an object into a JSON string.
    - A `fail` function that takes an error message as an argument and
      raises an exception on Python side.
    """

    def __init__(self, prompt_dirs: Sequence[Path], data_dirs: Sequence[Path]):
        """
        Args:
            prompt_dirs: A sequence of directories where Jinja prompt
                templates can be found.
            data_dirs: A sequence of directories where data files can be
                found.
        """
        self.prompt_folders = prompt_dirs
        self.data = _load_data(data_dirs)
        loader = jinja2.ChoiceLoader(
            [
                jinja2.FileSystemLoader(self.prompt_folders),
                jinja2.PackageLoader("delphyne.stdlib"),
            ]
        )
        self.env = jinja2.Environment(
            loader=loader,
            trim_blocks=True,
            lstrip_blocks=True,
            keep_trailing_newline=False,
        )
        self.env.filters["yaml"] = dump_yaml_object
        self.env.filters["json"] = _dump_json_object
        self.env.globals["fail"] = _fail_from_template  # type: ignore

    @override
    def prompt(
        self,
        *,
        query_name: str,
        prompt_kind: Literal["system", "instance"] | str,
        template_args: dict[str, Any],
        default_template: str | None = None,
    ) -> str:
        suffix = "." + prompt_kind
        template_name = f"{query_name}{suffix}{JINJA_EXTENSION}"
        try:
            template = self.env.get_template(template_name)
        except jinja2.TemplateNotFound:
            if default_template is not None:
                template = self.env.from_string(default_template)
            else:
                raise qu.TemplateFileMissing(template_name)
        try:
            assert "data" not in template_args
            template_args |= {"data": self.data}
            return template.render(template_args)
        except jinja2.TemplateNotFound as e:
            raise qu.TemplateError(template_name, e)
        except jinja2.UndefinedError as e:
            raise qu.TemplateError(template_name, e)
        except jinja2.TemplateSyntaxError as e:
            raise qu.TemplateError(template_name, e)

init

__init__(prompt_dirs: Sequence[Path], data_dirs: Sequence[Path])

Parameters:

Name	Type	Description	Default
`prompt_dirs`	`Sequence[Path]`	A sequence of directories where Jinja prompt templates can be found.	required
`data_dirs`	`Sequence[Path]`	A sequence of directories where data files can be found.	required

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

def __init__(self, prompt_dirs: Sequence[Path], data_dirs: Sequence[Path]):
    """
    Args:
        prompt_dirs: A sequence of directories where Jinja prompt
            templates can be found.
        data_dirs: A sequence of directories where data files can be
            found.
    """
    self.prompt_folders = prompt_dirs
    self.data = _load_data(data_dirs)
    loader = jinja2.ChoiceLoader(
        [
            jinja2.FileSystemLoader(self.prompt_folders),
            jinja2.PackageLoader("delphyne.stdlib"),
        ]
    )
    self.env = jinja2.Environment(
        loader=loader,
        trim_blocks=True,
        lstrip_blocks=True,
        keep_trailing_newline=False,
    )
    self.env.filters["yaml"] = dump_yaml_object
    self.env.filters["json"] = _dump_json_object
    self.env.globals["fail"] = _fail_from_template  # type: ignore

TemplateFileMissing `dataclass`

Bases: Exception

Exception raised when a template file is missing.

This exception should only be raised when a top-level template file is missing. If an include statement fails within a template, a TemplateError exception should be raised instead.

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

@dataclass
class TemplateFileMissing(Exception):
    """
    Exception raised when a template file is missing.

    This exception should only be raised when a top-level template file
    is missing. If an `include` statement fails within a template, a
    `TemplateError` exception should be raised instead.
    """

    file: str

Policy Environments

Main Class

PolicyEnv

__init__

InvalidDemoFile dataclass

Example Database

Example dataclass

QuerySerializedArgs

ExampleDatabase dataclass

add_query_demonstration

add_demonstration

examples

Tracer

TemplatesManager

__init__

TemplateFileMissing dataclass

init

InvalidDemoFile `dataclass`

Example `dataclass`

ExampleDatabase `dataclass`

init

TemplateFileMissing `dataclass`