orchestrata_ultimate.providers.adapters.ollama_adapter

Reference › orchestrata_ultimate › providers › adapters › ollama_adapter

ollama_adapter is the adapter for a local Ollama server (specification 011, requirement 033). OllamaAdapter extends BaseAdapter (base) and implements complete, get_models and supports_capability. It needs no API key, and client registers it without checking credentials; the budget profile uses it for local inference.

At a glance

Class diagram

Dependency diagram

API

Classes

OllamaAdapter

Bases: BaseAdapter · Decorators: traces(spec='011', story='US-001', reqs=['REQ-011-033']) · Line 22

Ollama local API provider adapter.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/adapters/ollama_adapter.py

"""Ollama provider adapter.

SPEC-011 FR-007: Implements IProvider for local Ollama API.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    Message,
    ModelCapability,
    ModelInfo,
    RawProviderResponse,
)
from orchestrata_ultimate.providers.adapters.base import BaseAdapter

logger = logging.getLogger(__name__)


@traces(spec="011", story="US-001", reqs=["REQ-011-033"])
class OllamaAdapter(BaseAdapter):
    """Ollama local API provider adapter."""

    def _get_headers(self) -> dict[str, str]:
        return {"Content-Type": "application/json"}

    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to Ollama."""
        client = await self._ensure_client()
        payload = {
            "model": model,
            "messages": [{"role": m.role, "content": m.content} for m in messages],
            "stream": False,
            **{k: v for k, v in parameters.items() if k != "stream"},
        }
        response = await client.post("/api/chat", json=payload)
        response.raise_for_status()
        data = response.json()
        message = data.get("message", {})
        # Ollama provides token counts in response
        prompt_eval_count = data.get("prompt_eval_count", 0)
        eval_count = data.get("eval_count", 0)
        return RawProviderResponse(
            content=message.get("content", ""),
            input_tokens=prompt_eval_count,
            output_tokens=eval_count,
            model=data.get("model", model),
            metadata={
                "total_duration": data.get("total_duration", 0),
                "load_duration": data.get("load_duration", 0),
            },
        )

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models."""
        return [
            ModelInfo(
                name=model_name,
                provider="ollama",
                capabilities=pricing.capabilities,
                context_window=pricing.context_window,
                pricing=pricing,
            )
            for model_name, pricing in self._config.pricing.items()
        ]

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if any model supports the capability."""
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False