Skip to content

Engine

Engine

Bases: Generic[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]

Source code in sieves/engines/core.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class Engine(Generic[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]):
    def __init__(
        self,
        model: EngineModel,
        init_kwargs: dict[str, Any] | None = None,
        inference_kwargs: dict[str, Any] | None = None,
        strict_mode: bool = False,
        batch_size: int = -1,
    ):
        """
        :param model: Instantiated model instance.
        :param init_kwargs: Optional kwargs to supply to engine executable at init time.
        :param inference_kwargs: Optional kwargs to supply to engine executable at inference time.
        :param strict_mode: If True, exception is raised if prompt response can't be parsed correctly.
        :param batch_size: Batch size in processing prompts. -1 will batch all documents in one go. Not all engines
            support batching.
        """
        self._model = model
        self._inference_kwargs = inference_kwargs or {}
        self._init_kwargs = init_kwargs or {}
        self._strict_mode = strict_mode
        self._batch_size = self._validate_batch_size(batch_size)

    def _validate_batch_size(self, batch_size: int) -> int:
        """Validates batch_size. Noop by default.
        :param batch_size: Specified batch size.
        :returns int: Validated batch size.
        """
        return batch_size

    @property
    def model(self) -> EngineModel:
        """Return model instance.
        :return: Model instance.
        """
        return self._model

    @property
    @abc.abstractmethod
    def supports_few_shotting(self) -> bool:
        """Whether engine supports few-shotting. If not, only zero-shotting is supported.
        :return: Whether engine supports few-shotting.
        """

    @property
    @abc.abstractmethod
    def inference_modes(self) -> type[EngineInferenceMode]:
        """Which inference modes are supported.
        :return: Supported inference modes.
        """

    @abc.abstractmethod
    def build_executable(
        self,
        inference_mode: EngineInferenceMode,
        prompt_template: str | None,
        prompt_signature: type[EnginePromptSignature] | EnginePromptSignature,
        fewshot_examples: Iterable[pydantic.BaseModel] = (),
    ) -> Executable[EngineResult | None]:
        """
        Returns prompt executable, i.e. a function that wraps an engine-native prediction generators. Such engine-native
        generators are e.g. Predict in DSPy, generator in outlines, Jsonformer in jsonformers).
        :param inference_mode: Inference mode to use (e.g. classification, JSON, ... - this is engine-specific).
        :param prompt_template: Prompt template.
        :param prompt_signature: Expected prompt signature type.
        :param fewshot_examples: Few-shot examples.
        :return: Prompt executable.
        """

    @staticmethod
    def _convert_fewshot_examples(fewshot_examples: Iterable[pydantic.BaseModel]) -> list[dict[str, Any]]:
        """
        Convert fewshot examples from pydantic.BaseModel instance to dicts.
        :param fewshot_examples: Fewshot examples to convert.
        :return: Fewshot examples as dicts.
        """
        return [fs_example.model_dump(serialize_as_any=True) for fs_example in fewshot_examples]

    @property
    def _attributes(self) -> dict[str, Attribute]:
        """Returns attributes to serialize.
        :return: Dict of attributes to serialize.
        """
        # Note: init_kwargs and inference_kwargs are potentially unfit for serialization as they contain arbitrary
        # objects.
        return {
            "model": Attribute(value=self._model),
            "init_kwargs": Attribute(value=self._init_kwargs),
            "inference_kwargs": Attribute(value=self._inference_kwargs),
        }

    def serialize(self) -> Config:
        """Serializes engine.
        :return: Config instance.
        """
        return Config.create(self.__class__, self._attributes)

    @classmethod
    def deserialize(
        cls, config: Config, **kwargs: dict[str, Any]
    ) -> Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]:
        """Generate Engine instance from config.
        :param config: Config to generate instance from.
        :param kwargs: Values to inject into loaded config.
        :return: Deserialized Engine instance.
        """
        return cls(**config.to_init_dict(cls, **kwargs))

    @staticmethod
    async def _execute_async_calls(calls: list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]) -> Any:
        """Executes batch of async functions.
        :param calls: Async calls to execute.
        :return: Parsed response objects.
        """
        return await asyncio.gather(*calls)

_attributes property

Returns attributes to serialize.

Returns:

Type Description
dict[str, Attribute]

Dict of attributes to serialize.

inference_modes abstractmethod property

Which inference modes are supported.

Returns:

Type Description
type[EngineInferenceMode]

Supported inference modes.

model property

Return model instance.

Returns:

Type Description
EngineModel

Model instance.

supports_few_shotting abstractmethod property

Whether engine supports few-shotting. If not, only zero-shotting is supported.

Returns:

Type Description
bool

Whether engine supports few-shotting.

__init__(model, init_kwargs=None, inference_kwargs=None, strict_mode=False, batch_size=-1)

Parameters:

Name Type Description Default
model EngineModel

Instantiated model instance.

 required
init_kwargs dict[str, Any] | None

Optional kwargs to supply to engine executable at init time.

 None
inference_kwargs dict[str, Any] | None

Optional kwargs to supply to engine executable at inference time.

 None
strict_mode bool

If True, exception is raised if prompt response can't be parsed correctly.

 False
batch_size int

Batch size in processing prompts. -1 will batch all documents in one go. Not all engines support batching.

 -1
Source code in sieves/engines/core.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    model: EngineModel,
    init_kwargs: dict[str, Any] | None = None,
    inference_kwargs: dict[str, Any] | None = None,
    strict_mode: bool = False,
    batch_size: int = -1,
):
    """
    :param model: Instantiated model instance.
    :param init_kwargs: Optional kwargs to supply to engine executable at init time.
    :param inference_kwargs: Optional kwargs to supply to engine executable at inference time.
    :param strict_mode: If True, exception is raised if prompt response can't be parsed correctly.
    :param batch_size: Batch size in processing prompts. -1 will batch all documents in one go. Not all engines
        support batching.
    """
    self._model = model
    self._inference_kwargs = inference_kwargs or {}
    self._init_kwargs = init_kwargs or {}
    self._strict_mode = strict_mode
    self._batch_size = self._validate_batch_size(batch_size)

_convert_fewshot_examples(fewshot_examples) staticmethod

Convert fewshot examples from pydantic.BaseModel instance to dicts.

Parameters:

Name Type Description Default
fewshot_examples Iterable[BaseModel]

Fewshot examples to convert.

 required

Returns:

Type Description
list[dict[str, Any]]

Fewshot examples as dicts.
Source code in sieves/engines/core.py 
 96
 97
 98
 99
100
101
102
103
@staticmethod
def _convert_fewshot_examples(fewshot_examples: Iterable[pydantic.BaseModel]) -> list[dict[str, Any]]:
    """
    Convert fewshot examples from pydantic.BaseModel instance to dicts.
    :param fewshot_examples: Fewshot examples to convert.
    :return: Fewshot examples as dicts.
    """
    return [fs_example.model_dump(serialize_as_any=True) for fs_example in fewshot_examples]

_execute_async_calls(calls) async staticmethod

Executes batch of async functions.

Parameters:

Name Type Description Default
calls list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]

Async calls to execute.

 required

Returns:

Type Description
Any

Parsed response objects.
Source code in sieves/engines/core.py 
135
136
137
138
139
140
141
@staticmethod
async def _execute_async_calls(calls: list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]) -> Any:
    """Executes batch of async functions.
    :param calls: Async calls to execute.
    :return: Parsed response objects.
    """
    return await asyncio.gather(*calls)

_validate_batch_size(batch_size)

Validates batch_size. Noop by default.

Parameters:

Name Type Description Default
batch_size int

Specified batch size.

 required

Returns:

Type Description
int

Validated batch size.
Source code in sieves/engines/core.py 
50
51
52
53
54
55
def _validate_batch_size(self, batch_size: int) -> int:
    """Validates batch_size. Noop by default.
    :param batch_size: Specified batch size.
    :returns int: Validated batch size.
    """
    return batch_size

build_executable(inference_mode, prompt_template, prompt_signature, fewshot_examples=()) abstractmethod

Returns prompt executable, i.e. a function that wraps an engine-native prediction generators. Such engine-native generators are e.g. Predict in DSPy, generator in outlines, Jsonformer in jsonformers).

Parameters:

Name Type Description Default
inference_mode EngineInferenceMode

Inference mode to use (e.g. classification, JSON, ... - this is engine-specific).

 required
prompt_template str | None

Prompt template.

 required
prompt_signature type[EnginePromptSignature] | EnginePromptSignature

Expected prompt signature type.

 required
fewshot_examples Iterable[BaseModel]

Few-shot examples.

 ()

Returns:

Type Description
Executable[EngineResult | None]

Prompt executable.
Source code in sieves/engines/core.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@abc.abstractmethod
def build_executable(
    self,
    inference_mode: EngineInferenceMode,
    prompt_template: str | None,
    prompt_signature: type[EnginePromptSignature] | EnginePromptSignature,
    fewshot_examples: Iterable[pydantic.BaseModel] = (),
) -> Executable[EngineResult | None]:
    """
    Returns prompt executable, i.e. a function that wraps an engine-native prediction generators. Such engine-native
    generators are e.g. Predict in DSPy, generator in outlines, Jsonformer in jsonformers).
    :param inference_mode: Inference mode to use (e.g. classification, JSON, ... - this is engine-specific).
    :param prompt_template: Prompt template.
    :param prompt_signature: Expected prompt signature type.
    :param fewshot_examples: Few-shot examples.
    :return: Prompt executable.
    """

deserialize(config, **kwargs) classmethod

Generate Engine instance from config.

Parameters:

Name Type Description Default
config Config

Config to generate instance from.

 required
kwargs dict[str, Any]

Values to inject into loaded config.

 {}

Returns:

Type Description
Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]

Deserialized Engine instance.
Source code in sieves/engines/core.py 
124
125
126
127
128
129
130
131
132
133
@classmethod
def deserialize(
    cls, config: Config, **kwargs: dict[str, Any]
) -> Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]:
    """Generate Engine instance from config.
    :param config: Config to generate instance from.
    :param kwargs: Values to inject into loaded config.
    :return: Deserialized Engine instance.
    """
    return cls(**config.to_init_dict(cls, **kwargs))

serialize()

Serializes engine.

Returns:

Type Description
Config

Config instance.
Source code in sieves/engines/core.py 
118
119
120
121
122
def serialize(self) -> Config:
    """Serializes engine.
    :return: Config instance.
    """
    return Config.create(self.__class__, self._attributes)

PydanticEngine

Bases: ABC, Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]

Abstract super class for all engines working directly with Pydantic objects for prompt signatures and results. Note that this class also assumes the engine accepts a prompt. This holds true for most engines - it doesn't only for those with an idiocratic way to process prompts like DSPy, or decoder-only models which don't work with object-based signatures anyway. If and once we add support for a Pydantic-based engine that doesn't accept prompt templates, we'll adjust by modifying _infer() to accept an additional parameter specifying how to handle prompt/instruction injection (and we might have to make supports_few_shotting() engine-specific again).

Source code in sieves/engines/core.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
class PydanticEngine(abc.ABC, Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]):
    """Abstract super class for all engines working directly with Pydantic objects for prompt signatures and results.
    Note that this class also assumes the engine accepts a prompt. This holds true for most engines - it doesn't only
    for those with an idiocratic way to process prompts like DSPy, or decoder-only models which don't work with
    object-based signatures anyway.
    If and once we add support for a Pydantic-based engine that doesn't accept prompt templates, we'll adjust by
    modifying `_infer()` to accept an additional parameter specifying how to handle prompt/instruction injection (and
    we might have to make `supports_few_shotting()` engine-specific again).
    """

    @classmethod
    def _create_template(cls, template: str | None) -> jinja2.Template:
        """Creates Jinja2 template from template string.
        :param template: Template string.
        :return: Jinja2 template.
        """
        assert template, f"prompt_template has to be provided to {cls.__name__}."
        return jinja2.Template(template)

    @property
    def supports_few_shotting(self) -> bool:
        return True

    def _infer(
        self,
        generator: Callable[[list[str]], Iterable[EngineResult]],
        template: jinja2.Template,
        values: Iterable[dict[str, Any]],
        fewshot_examples: Iterable[pydantic.BaseModel],
    ) -> Iterable[EngineResult | None]:
        """
        Runs inference record by record with exception handling for template- and Pydantic-based engines.
        :param generator: Callable generating responses.
        :param template: Prompt template.
        :param values: Doc values to inject.
        :param fewshot_examples: Fewshot examples.
        :return: Results parsed from responses.
        """
        fewshot_examples_dict = Engine._convert_fewshot_examples(fewshot_examples)
        examples = {"examples": fewshot_examples_dict} if len(fewshot_examples_dict) else {}
        batch_size = self._batch_size if self._batch_size != -1 else sys.maxsize
        # Ensure values are read as generator for standardized batch handling (otherwise we'd have to use different
        # batch handling depending on whether lists/tuples or generators are used).
        values = (v for v in values)

        while batch := [vals for vals in itertools.islice(values, batch_size)]:
            if len(batch) == 0:
                break

            try:
                yield from generator([template.render(**doc_values, **examples) for doc_values in batch])

            except (TypeError, pydantic.ValidationError) as err:
                if self._strict_mode:
                    raise ValueError(
                        "Encountered problem when executing prompt. Ensure your few-shot examples and document "
                        "chunks contain sensible information."
                    ) from err
                else:
                    yield from (None for _ in range(len(batch)))

_attributes property

Returns attributes to serialize.

Returns:

Type Description
dict[str, Attribute]

Dict of attributes to serialize.

inference_modes abstractmethod property

Which inference modes are supported.

Returns:

Type Description
type[EngineInferenceMode]

Supported inference modes.

model property

Return model instance.

Returns:

Type Description
EngineModel

Model instance.

__init__(model, init_kwargs=None, inference_kwargs=None, strict_mode=False, batch_size=-1)

Parameters:

Name Type Description Default
model EngineModel

Instantiated model instance.

 required
init_kwargs dict[str, Any] | None

Optional kwargs to supply to engine executable at init time.

 None
inference_kwargs dict[str, Any] | None

Optional kwargs to supply to engine executable at inference time.

 None
strict_mode bool

If True, exception is raised if prompt response can't be parsed correctly.

 False
batch_size int

Batch size in processing prompts. -1 will batch all documents in one go. Not all engines support batching.

 -1
Source code in sieves/engines/core.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    model: EngineModel,
    init_kwargs: dict[str, Any] | None = None,
    inference_kwargs: dict[str, Any] | None = None,
    strict_mode: bool = False,
    batch_size: int = -1,
):
    """
    :param model: Instantiated model instance.
    :param init_kwargs: Optional kwargs to supply to engine executable at init time.
    :param inference_kwargs: Optional kwargs to supply to engine executable at inference time.
    :param strict_mode: If True, exception is raised if prompt response can't be parsed correctly.
    :param batch_size: Batch size in processing prompts. -1 will batch all documents in one go. Not all engines
        support batching.
    """
    self._model = model
    self._inference_kwargs = inference_kwargs or {}
    self._init_kwargs = init_kwargs or {}
    self._strict_mode = strict_mode
    self._batch_size = self._validate_batch_size(batch_size)

_convert_fewshot_examples(fewshot_examples) staticmethod

Convert fewshot examples from pydantic.BaseModel instance to dicts.

Parameters:

Name Type Description Default
fewshot_examples Iterable[BaseModel]

Fewshot examples to convert.

 required

Returns:

Type Description
list[dict[str, Any]]

Fewshot examples as dicts.
Source code in sieves/engines/core.py 
 96
 97
 98
 99
100
101
102
103
@staticmethod
def _convert_fewshot_examples(fewshot_examples: Iterable[pydantic.BaseModel]) -> list[dict[str, Any]]:
    """
    Convert fewshot examples from pydantic.BaseModel instance to dicts.
    :param fewshot_examples: Fewshot examples to convert.
    :return: Fewshot examples as dicts.
    """
    return [fs_example.model_dump(serialize_as_any=True) for fs_example in fewshot_examples]

_create_template(template) classmethod

Creates Jinja2 template from template string.

Parameters:

Name Type Description Default
template str | None

Template string.

 required

Returns:

Type Description
Template

Jinja2 template.
Source code in sieves/engines/core.py 
154
155
156
157
158
159
160
161
@classmethod
def _create_template(cls, template: str | None) -> jinja2.Template:
    """Creates Jinja2 template from template string.
    :param template: Template string.
    :return: Jinja2 template.
    """
    assert template, f"prompt_template has to be provided to {cls.__name__}."
    return jinja2.Template(template)

_execute_async_calls(calls) async staticmethod

Executes batch of async functions.

Parameters:

Name Type Description Default
calls list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]

Async calls to execute.

 required

Returns:

Type Description
Any

Parsed response objects.
Source code in sieves/engines/core.py 
135
136
137
138
139
140
141
@staticmethod
async def _execute_async_calls(calls: list[Coroutine[Any, Any, Any]] | list[Awaitable[Any]]) -> Any:
    """Executes batch of async functions.
    :param calls: Async calls to execute.
    :return: Parsed response objects.
    """
    return await asyncio.gather(*calls)

_infer(generator, template, values, fewshot_examples)

Runs inference record by record with exception handling for template- and Pydantic-based engines.

Parameters:

Name Type Description Default
generator Callable[[list[str]], Iterable[EngineResult]]

Callable generating responses.

 required
template Template

Prompt template.

 required
values Iterable[dict[str, Any]]

Doc values to inject.

 required
fewshot_examples Iterable[BaseModel]

Fewshot examples.

 required

Returns:

Type Description
Iterable[EngineResult | None]

Results parsed from responses.
Source code in sieves/engines/core.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def _infer(
    self,
    generator: Callable[[list[str]], Iterable[EngineResult]],
    template: jinja2.Template,
    values: Iterable[dict[str, Any]],
    fewshot_examples: Iterable[pydantic.BaseModel],
) -> Iterable[EngineResult | None]:
    """
    Runs inference record by record with exception handling for template- and Pydantic-based engines.
    :param generator: Callable generating responses.
    :param template: Prompt template.
    :param values: Doc values to inject.
    :param fewshot_examples: Fewshot examples.
    :return: Results parsed from responses.
    """
    fewshot_examples_dict = Engine._convert_fewshot_examples(fewshot_examples)
    examples = {"examples": fewshot_examples_dict} if len(fewshot_examples_dict) else {}
    batch_size = self._batch_size if self._batch_size != -1 else sys.maxsize
    # Ensure values are read as generator for standardized batch handling (otherwise we'd have to use different
    # batch handling depending on whether lists/tuples or generators are used).
    values = (v for v in values)

    while batch := [vals for vals in itertools.islice(values, batch_size)]:
        if len(batch) == 0:
            break

        try:
            yield from generator([template.render(**doc_values, **examples) for doc_values in batch])

        except (TypeError, pydantic.ValidationError) as err:
            if self._strict_mode:
                raise ValueError(
                    "Encountered problem when executing prompt. Ensure your few-shot examples and document "
                    "chunks contain sensible information."
                ) from err
            else:
                yield from (None for _ in range(len(batch)))

_validate_batch_size(batch_size)

Validates batch_size. Noop by default.

Parameters:

Name Type Description Default
batch_size int

Specified batch size.

 required

Returns:

Type Description
int

Validated batch size.
Source code in sieves/engines/core.py 
50
51
52
53
54
55
def _validate_batch_size(self, batch_size: int) -> int:
    """Validates batch_size. Noop by default.
    :param batch_size: Specified batch size.
    :returns int: Validated batch size.
    """
    return batch_size

build_executable(inference_mode, prompt_template, prompt_signature, fewshot_examples=()) abstractmethod

Returns prompt executable, i.e. a function that wraps an engine-native prediction generators. Such engine-native generators are e.g. Predict in DSPy, generator in outlines, Jsonformer in jsonformers).

Parameters:

Name Type Description Default
inference_mode EngineInferenceMode

Inference mode to use (e.g. classification, JSON, ... - this is engine-specific).

 required
prompt_template str | None

Prompt template.

 required
prompt_signature type[EnginePromptSignature] | EnginePromptSignature

Expected prompt signature type.

 required
fewshot_examples Iterable[BaseModel]

Few-shot examples.

 ()

Returns:

Type Description
Executable[EngineResult | None]

Prompt executable.
Source code in sieves/engines/core.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@abc.abstractmethod
def build_executable(
    self,
    inference_mode: EngineInferenceMode,
    prompt_template: str | None,
    prompt_signature: type[EnginePromptSignature] | EnginePromptSignature,
    fewshot_examples: Iterable[pydantic.BaseModel] = (),
) -> Executable[EngineResult | None]:
    """
    Returns prompt executable, i.e. a function that wraps an engine-native prediction generators. Such engine-native
    generators are e.g. Predict in DSPy, generator in outlines, Jsonformer in jsonformers).
    :param inference_mode: Inference mode to use (e.g. classification, JSON, ... - this is engine-specific).
    :param prompt_template: Prompt template.
    :param prompt_signature: Expected prompt signature type.
    :param fewshot_examples: Few-shot examples.
    :return: Prompt executable.
    """

deserialize(config, **kwargs) classmethod

Generate Engine instance from config.

Parameters:

Name Type Description Default
config Config

Config to generate instance from.

 required
kwargs dict[str, Any]

Values to inject into loaded config.

 {}

Returns:

Type Description
Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]

Deserialized Engine instance.
Source code in sieves/engines/core.py 
124
125
126
127
128
129
130
131
132
133
@classmethod
def deserialize(
    cls, config: Config, **kwargs: dict[str, Any]
) -> Engine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]:
    """Generate Engine instance from config.
    :param config: Config to generate instance from.
    :param kwargs: Values to inject into loaded config.
    :return: Deserialized Engine instance.
    """
    return cls(**config.to_init_dict(cls, **kwargs))

serialize()

Serializes engine.

Returns:

Type Description
Config

Config instance.
Source code in sieves/engines/core.py 
118
119
120
121
122
def serialize(self) -> Config:
    """Serializes engine.
    :return: Config instance.
    """
    return Config.create(self.__class__, self._attributes)