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(host=self._model.host, **({"timeout": 30} | self._model.client_config))

    @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.
            :raises: pydantic.ValidationError is response can't be parsed.
            :raises: httpx.ReadTimeout if request times out.
            """
            match inference_mode:
                case InferenceMode.structured:

                    def generate(prompts: list[str]) -> Iterable[Result]:
                        responses: list[Any] | None = None
                        n_tries = 0

                        while responses is None:
                            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
                            ]

                            try:
                                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 "
                                        f"prompts and examples."
                                    ) from ex

                            except (RuntimeError, httpx.ReadTimeout) as err:
                                n_tries += 1
                                if n_tries > self._model.max_retries:
                                    raise TimeoutError("Hit timeout limit with Ollama.") from err

                                self._client = ollama.AsyncClient(
                                    host=self._model.host,
                                    **({"timeout": self._model.timeout} | self._model.client_config),
                                )

                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

model property

Return model instance.

Returns:

Type	Description
EngineModel	Model instance.

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
InternalEngine[EnginePromptSignature, EngineResult, EngineModel, EngineInferenceMode]	Deserialized Engine instance.

Source code in sieves/engines/core.py

@classmethod
def deserialize(
    cls, config: Config, **kwargs: dict[str, Any]
) -> InternalEngine[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)