Ollama

Bases: PydanticEngine[PromptSignature, Result, Model, InferenceMode]

Engine for Ollama. Make sure a Ollama server is running. In a nutshell: > curl -fsSL https://ollama.ai/install.sh | sh > ollama serve (or ollama run MODEL_ID)

Source code in sieves/engines/ollama_.py

class Ollama(PydanticEngine[PromptSignature, Result, Model, InferenceMode]):
    """Engine for Ollama.
    Make sure a Ollama server is running.
            In a nutshell:
            > curl -fsSL https://ollama.ai/install.sh | sh
            > ollama serve (or ollama run MODEL_ID)
    """

    def __init__(
        self,
        model: Model,
        init_kwargs: dict[str, Any] | None = None,
        inference_kwargs: dict[str, Any] | None = None,
        strict_mode: bool = False,
        batch_size: int = -1,
    ):
        super().__init__(
            model=model,
            init_kwargs=init_kwargs,
            inference_kwargs=inference_kwargs,
            strict_mode=strict_mode,
            batch_size=batch_size,
        )
        # Async client will be initialized for every prompt batch to sidestep an asyncio event loop issue.
        self._client: ollama.AsyncClient | None = None

    @property
    def inference_modes(self) -> type[InferenceMode]:
        return InferenceMode

    def build_executable(
        self,
        inference_mode: InferenceMode,
        prompt_template: str | None,  # noqa: UP007
        prompt_signature: type[PromptSignature] | PromptSignature,
        fewshot_examples: Iterable[pydantic.BaseModel] = tuple(),
    ) -> Executable[Result | None]:
        assert isinstance(prompt_signature, type)
        cls_name = self.__class__.__name__
        template = self._create_template(prompt_template)

        def execute(values: Iterable[dict[str, Any]]) -> Iterable[Result | None]:
            """Execute prompts with engine for given values.
            :param values: Values to inject into prompts.
            :return: Results for prompts. Results are None if corresponding prompt failed.
            """
            match inference_mode:
                case InferenceMode.chat:

                    def generate(prompts: list[str]) -> Iterable[Result]:
                        self._client = ollama.AsyncClient(host=self._model.host, **self._model.client_config)
                        assert self._client

                        calls = [
                            self._client.chat(
                                messages=[{"role": "user", "content": prompt}],
                                model=self._model.name,
                                format=prompt_signature.model_json_schema(),
                                **self._inference_kwargs,
                            )
                            for prompt in prompts
                        ]
                        responses = asyncio.run(self._execute_async_calls(calls))

                        try:
                            for res in responses:
                                yield prompt_signature.model_validate_json(res.message.content)
                        except pydantic.ValidationError as ex:
                            raise pydantic.ValidationError(
                                f"Encountered problem in parsing {cls_name} output. Double-check your prompts and "
                                f"examples."
                            ) from ex

                case _:
                    raise ValueError(f"Inference mode {inference_mode} not supported by {cls_name} engine.")

            yield from self._infer(generate, template, values, fewshot_examples)

        return execute

_attributes property

Returns attributes to serialize.

Returns:

Type	Description
dict[str, Attribute]	Dict of attributes to serialize.

model property

Return model instance.

Returns:

Type	Description
EngineModel	Model instance.

_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

@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

@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

@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

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

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

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

@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

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