OllamaEngine(
host: str | None = None,
*,
timeout: float = 300.0,
keep_alive: str = "30m",
)
Bases: AsyncHTTPEngineMixin, InferenceEngine
Ollama backend via its native HTTP API.
Source code in src/diapason/engine/ollama.py
| def __init__(
self,
host: str | None = None,
*,
timeout: float = 300.0,
keep_alive: str = "30m",
) -> None:
# Priority: explicit host (from config.toml) > OLLAMA_HOST env var > default
if host is None:
env_host = os.environ.get("OLLAMA_HOST")
host = env_host or self._DEFAULT_HOST
self._host = host.rstrip("/")
self._scheduler = scheduler_for(self._host)
# Used by the shared async streaming plumbing (AsyncHTTPEngineMixin) so a
# wedged token read is bounded by ``timeout`` instead of hanging the
# single event loop for the httpx default.
self._timeout = timeout
self._keep_alive = (keep_alive or "30m").strip()
# Injection seam for tests: an ``httpx.MockTransport`` swapped in here drives
# the async stream path with no real Ollama server. ``None`` in production so
# httpx uses its default networking.
self._async_transport: httpx.AsyncBaseTransport | None = None
self._client = httpx.Client(base_url=self._host, timeout=timeout)
# Last stream usage — captured from Ollama's final chunk
self._last_stream_usage: Dict[str, int] = {}
|