windlass.providers.llm.ollama

ollama

Ollama adapter for locally hosted models.

Ollama needs no extra install: it speaks HTTP and Windlass already depends on httpx. Start the daemon, pull a model, and point Windlass at it::

ollama pull llama3.2
python -c "from windlass import Windlass; print(Windlass.llm('ollama').complete('hi'))"

That makes Ollama the zero-cost, zero-key way to run the whole framework against a real model — useful for local development and for CI that wants more than the fake provider.

Example

from windlass import Windlass # doctest: +SKIP Windlass.llm("ollama", model="llama3.2").complete("hi").content # doctest: +SKIP

OllamaLLM

OllamaLLM(
    model: str = "",
    *,
    base_url: str | None = None,
    keep_alive: str | None = None,
    options: dict[str, Any] | None = None,
    **config: Any
)

Bases: LLM

Chat completions against a local Ollama daemon.

Parameters:

Name	Type	Description	Default
model	str	Model tag, e.g. llama3.2, qwen2.5, mistral.	''
base_url	str | None	Daemon address. Defaults to OLLAMA_BASE_URL or http://localhost:11434.	None
keep_alive	str | None	How long the daemon keeps the model resident, e.g. "5m".	None
options	dict[str, Any] | None	Raw Ollama options (num_ctx, top_k, repeat_penalty, ...), merged into the request.	None
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Raises:

Type	Description
ProviderError	When the daemon is unreachable. The message explains how to start it.

Performance

The first call after a cold start pays model-load time — often several seconds. Set keep_alive to keep the model warm between requests.

Source code in src\windlass\providers\llm\ollama.py

def __init__(
    self,
    model: str = "",
    *,
    base_url: str | None = None,
    keep_alive: str | None = None,
    options: dict[str, Any] | None = None,
    **config: Any,
) -> None:
    super().__init__(model=model, **config)
    self.base_url = (base_url or settings().ollama_base_url).rstrip("/")
    self.keep_alive = keep_alive
    self.options = dict(options or {})
    self._client = httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout)

default_model classmethod

default_model() -> str

Return "llama3.2".

Source code in src\windlass\providers\llm\ollama.py

@classmethod
def default_model(cls) -> str:
    """Return ``"llama3.2"``."""
    return "llama3.2"

native

native() -> Any

Return the underlying httpx.AsyncClient.

Source code in src\windlass\providers\llm\ollama.py

def native(self) -> Any:
    """Return the underlying ``httpx.AsyncClient``."""
    return self._client

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Request one chat completion.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	OpenAI-format tool definitions. Support depends on the model.	None
**kwargs	Any	Request overrides.	{}

Returns:

Type	Description
Completion	The completion.

Raises:

Type	Description
ProviderError	When the daemon is unreachable or returns an error.
ProviderTimeoutError	When the request exceeds the timeout.

Source code in src\windlass\providers\llm\ollama.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Request one chat completion.

    Args:
        messages: The conversation.
        tools: OpenAI-format tool definitions. Support depends on the model.
        **kwargs: Request overrides.

    Returns:
        The completion.

    Raises:
        ProviderError: When the daemon is unreachable or returns an error.
        ProviderTimeoutError: When the request exceeds the timeout.
    """
    payload = self._payload(messages, tools, kwargs, stream=False)
    data = await self._post("/api/chat", payload)

    message = data.get("message") or {}
    calls = [
        ToolCall(
            name=(call.get("function") or {}).get("name", ""),
            arguments=_as_dict((call.get("function") or {}).get("arguments")),
        )
        for call in message.get("tool_calls") or []
    ]
    return Completion(
        content=message.get("content", ""),
        tool_calls=calls,
        finish_reason=FinishReason.TOOL_CALLS if calls else FinishReason.STOP,
        model=data.get("model", self.model),
        usage=Usage(
            prompt_tokens=data.get("prompt_eval_count", 0) or 0,
            completion_tokens=data.get("eval_count", 0) or 0,
        ),
        raw=data,
    )

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream a chat completion.

Ollama streams newline-delimited JSON objects rather than SSE.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	OpenAI-format tool definitions.	None
**kwargs	Any	Request overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Text deltas, then tool calls, then done.

Source code in src\windlass\providers\llm\ollama.py

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream a chat completion.

    Ollama streams newline-delimited JSON objects rather than SSE.

    Args:
        messages: The conversation.
        tools: OpenAI-format tool definitions.
        **kwargs: Request overrides.

    Yields:
        Text deltas, then tool calls, then ``done``.
    """
    payload = self._payload(messages, tools, kwargs, stream=True)
    calls: list[ToolCall] = []
    usage = Usage()

    try:
        async with self._client.stream("POST", "/api/chat", json=payload) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                if not line.strip():
                    continue
                try:
                    chunk = json.loads(line)
                except json.JSONDecodeError:
                    continue
                message = chunk.get("message") or {}
                if message.get("content"):
                    yield StreamEvent(type="text", delta=message["content"], raw=chunk)
                for call in message.get("tool_calls") or []:
                    fn = call.get("function") or {}
                    calls.append(
                        ToolCall(
                            name=fn.get("name", ""), arguments=_as_dict(fn.get("arguments"))
                        )
                    )
                if chunk.get("done"):
                    usage = Usage(
                        prompt_tokens=chunk.get("prompt_eval_count", 0) or 0,
                        completion_tokens=chunk.get("eval_count", 0) or 0,
                    )
    except httpx.HTTPError as exc:
        raise self._transport_error(exc) from exc

    for call in calls:
        yield StreamEvent(type="tool_call", tool_call=call)
    yield StreamEvent(
        type="done",
        finish_reason=FinishReason.TOOL_CALLS if calls else FinishReason.STOP,
        usage=usage,
    )

alist_models async

alist_models() -> list[str]

Return the model tags the daemon has pulled.

Returns:

Type	Description
list[str]	Model tags, e.g. ["llama3.2:latest", "nomic-embed-text:latest"].

Raises:

Type	Description
ProviderError	When the daemon is unreachable.

Source code in src\windlass\providers\llm\ollama.py

async def alist_models(self) -> list[str]:
    """Return the model tags the daemon has pulled.

    Returns:
        Model tags, e.g. ``["llama3.2:latest", "nomic-embed-text:latest"]``.

    Raises:
        ProviderError: When the daemon is unreachable.
    """
    try:
        response = await self._client.get("/api/tags")
        response.raise_for_status()
    except httpx.HTTPError as exc:
        raise self._transport_error(exc) from exc
    return [m["name"] for m in response.json().get("models", [])]

aclose async

aclose() -> None

Close the HTTP client.

Source code in src\windlass\providers\llm\ollama.py

async def aclose(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()