provider.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/ollama/provider.py
 
"""
OllamaProvider — :class:`ProviderPlugin` implementation for Ollama.
 
Purpose
-------
Connects OC Chat to a local (or LAN) Ollama daemon via the native
``/api/chat`` endpoint. No API key required. Base URL defaults to
``http://localhost:11434`` and is overridable via ``OLLAMA_BASE_URL``.
 
Responsibilities
----------------
- :class:`OllamaProvider` — implement every abstract method on
  :class:`ProviderPlugin` for the Ollama provider.
- :meth:`build_client` — return an :class:`OllamaClient`.
- :meth:`models` / :meth:`tier` — surface the static registries.
- :meth:`build_extra_body` — always ``None`` (no thinking payload).
- :meth:`transform_outgoing_messages` — pass-through (no sentinel).
- :meth:`parse_usage` — read ``prompt_eval_count`` / ``eval_count``
  from the Ollama-shaped usage dict yielded by :class:`OllamaClient`.
- :meth:`fetch_live_models` — delegate to ``client.list_models()``.
- :meth:`default_caps` — permissive local defaults for any live model
  not in the curated registry.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``api_key_env_var = ""`` — Ollama does not use API keys.
- ``transform_outgoing_messages`` is a true pass-through (no sentinel
  embedding — Ollama has no disk-based prefix cache that benefits from
  it).
- ``parse_usage`` handles both the Ollama-native field names
  (``prompt_eval_count`` / ``eval_count``) and the normalised names
  (``prompt_tokens`` / ``completion_tokens``) so it works regardless of
  which shape the caller passes.
 
Dependencies
------------
- oc_diagnostics (_dbg)
- providers.ollama.client (OllamaClient)
- providers.ollama.models (MODELS, TIER_MAP)
- option_c_agent_contracts (ProviderPlugin base class and the ModelCaps, ModelInfo, PerformanceStats, UsageStats and BaseClient types)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
from option_c_agent_contracts import (
    ProviderPlugin,
    ModelCaps,
    ModelInfo,
    PerformanceStats,
    UsageStats,
    BaseClient,
)
 
from .client import OllamaClient
from .models import MODELS, TIER_MAP
 
 
class OllamaProvider(ProviderPlugin):
    """Ollama LLM provider — native /api/chat, no API key."""
 
    name = "ollama"
    # OCT-LINT: disable=no_hardcoded_secrets reason="Empty env-var-name; Ollama uses no API key, not a credential"
    api_key_env_var = ""
    base_url_env_var = "OLLAMA_BASE_URL"
    default_base_url = "http://localhost:11434"
    default_model = "llama3.2"
 
    # ── Client construction ──────────────────────────────────────────────────
 
    def build_client(self, api_key: str, base_url: str) -> BaseClient:
        func = "build_client"
        _dbg("OC_CHAT.CLIENT", func, f"OllamaProvider base_url={base_url!r}", 3)
        return OllamaClient(api_key=api_key, base_url=base_url)
 
    # ── Model registry ───────────────────────────────────────────────────────
 
    def models(self) -> dict[str, ModelCaps]:
        return MODELS
 
    def fetch_live_models(self, client: BaseClient) -> list[str]:
        return client.list_models()
 
    def default_caps(self) -> ModelCaps:
        return ModelCaps(
            supports_thinking=False,
            supports_reasoning=False,
            supports_tools=True,
            max_context_tokens=8_192,
        )
 
    # ── Tier mapping ─────────────────────────────────────────────────────────
 
    def tier(self, name: str) -> str | None:
        return TIER_MAP.get(name)
 
    # ── Per-call shaping ─────────────────────────────────────────────────────
 
    def build_extra_body(
        self,
        model: str,
        *,
        enable_thinking: bool,
        effort: str | None,
    ) -> dict | None:
        return None  # Ollama has no extra_body concept
 
    def transform_outgoing_messages(
        self,
        messages: list[dict],
        tool_schemas: list[dict] | None,
    ) -> list[dict]:
        return messages  # Ollama uses the messages array as-is
 
    # ── Optional extension hooks ─────────────────────────────────────────────
 
    def translate_generation_options(self, opts: "GenerationOptions") -> dict | None:  # type: ignore[override]
        from option_c_agent_contracts import GenerationOptions
        d: dict = {}
        if opts.temperature is not None:    d["temperature"] = opts.temperature
        if opts.top_p is not None:          d["top_p"] = opts.top_p
        if opts.top_k is not None:          d["top_k"] = opts.top_k
        if opts.num_predict is not None:    d["num_predict"] = opts.num_predict
        if opts.repeat_penalty is not None: d["repeat_penalty"] = opts.repeat_penalty
        if opts.seed is not None:           d["seed"] = opts.seed
        return {"options": d} if d else None
 
    def parse_performance(self, chunk: dict) -> PerformanceStats:
        return PerformanceStats(
            prompt_eval_count=chunk.get("prompt_eval_count", 0),
            prompt_eval_duration_ns=chunk.get("prompt_eval_duration", 0),
            eval_count=chunk.get("eval_count", 0),
            eval_duration_ns=chunk.get("eval_duration", 0),
            total_duration_ns=chunk.get("total_duration", 0),
        )
 
    def model_info(self, model: str, client: BaseClient) -> ModelInfo | None:
        func = "model_info"
        try:
            data = client.show_model(model)  # type: ignore[attr-defined]
        except Exception as exc:
            _dbg("OC_CHAT.CLIENT", func, f"show_model failed: {exc}", 2)
            return None
        details = data.get("details") or {}
        return ModelInfo(
            name=model,
            family=details.get("family", ""),
            parameter_count=details.get("parameter_size", ""),
            quantization=details.get("quantization_level", ""),
            context_window=details.get("context_length", 0),
            raw_metadata=data,
        )
 
    # ── Response parsing ─────────────────────────────────────────────────────
 
    def parse_usage(self, usage: dict) -> UsageStats:
        if not isinstance(usage, dict):
            return UsageStats()
        # Accept both Ollama-native names and normalised names
        prompt = int(
            usage.get("prompt_eval_count")
            or usage.get("prompt_tokens")
            or 0
        )
        completion = int(
            usage.get("eval_count")
            or usage.get("completion_tokens")
            or 0
        )
        total = int(usage.get("total_tokens") or 0) or (prompt + completion)
        return UsageStats(
            prompt_tokens=prompt,
            completion_tokens=completion,
            total_tokens=total,
        )