mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
feat(cli): Ollama model discovery via probe (#3286)
Auto-populates the model switcher with locally installed Ollama models by probing the daemon lazily from the selector flow. `langchain-ollama` does not ship static profile data, so users previously had to curate `models = […]` and profile overrides by hand before local Ollama models showed useful metadata. ## Changes - **`get_available_models()`** merges models discovered from `GET /api/tags` with static profiles and explicit config lists, preserving config order and deduplicating across sources. - **`get_model_profiles()`** supplements Ollama models with best-effort metadata from `POST /api/show`, including context length and capabilities. Configured Ollama models are still inspected even when `/api/tags` returns no names. - **`_get_ollama_installed_models()`** caches successful `/api/tags` results per endpoint so selector loading does not probe tags separately for available models and profiles. Empty tag results are not cached, so a daemon that starts later can recover without `/reload`. - **`_fetch_ollama_installed_model_profiles()`** caches successful `/api/show` profile results per endpoint/model to avoid repeated serial profile probes while still allowing failed or empty results to retry. - **`_profile_from_ollama_show_payload()`** maps Ollama `*.context_length` values to `max_input_tokens`, maps `completion`, `tools`, and `thinking` capabilities to selector profile fields, and logs a debug canary when a shaped payload yields no recognized fields. - **`_fetch_ollama_installed_models()`** and **`_fetch_ollama_installed_model_profiles()`** share discovery headers, use a short timeout, and degrade gracefully on network or parse failures. Discovery bearer auth is only attached for local endpoints, so `OLLAMA_API_KEY` is not forwarded to arbitrary remote `base_url` values. - **`_ollama_discovery_enabled()`** keeps discovery enabled by default, accepts explicit truthy/falsy values through `DEEPAGENTS_CLI_OLLAMA_DISCOVERY`, and warns on unrecognized values. Config and CLI profile overrides continue to win over discovered Ollama metadata. The staged unit tests cover model listing, cached tag discovery, no negative caching for tag misses, successful profile caching, `/api/show` profile parsing, configured-model fallback, local-only discovery auth, malformed tag payloads, malformed JSON, provider opt-out cases, schema edge cases, cache clearing, and override precedence.
This commit is contained in:
Mason Daugherty
@@ -100,6 +100,15 @@ LANGSMITH_PROJECT = "DEEPAGENTS_CLI_LANGSMITH_PROJECT"
|
||||
NO_UPDATE_CHECK = "DEEPAGENTS_CLI_NO_UPDATE_CHECK"
|
||||
"""Disable automatic update checking when set."""
|
||||
|
||||
OLLAMA_DISCOVERY = "DEEPAGENTS_CLI_OLLAMA_DISCOVERY"
|
||||
"""Toggle Ollama model and profile discovery probes.
|
||||
|
||||
Defaults to enabled. Suppress the probe when the daemon is intentionally
|
||||
offline or the probe latency is undesirable. The probe is lazy and never
|
||||
runs on the startup hot path. When enabled, discovery may call `/api/tags`
|
||||
and `/api/show`. See `_ollama_discovery_enabled` for accepted truthy/falsy
|
||||
values."""
|
||||
|
||||
SERVER_ENV_PREFIX = "DEEPAGENTS_CLI_SERVER_"
|
||||
"""Environment variable prefix used to pass CLI config to the server subprocess."""
|
||||
|
||||
|
||||
@@ -17,12 +17,12 @@ from dataclasses import dataclass, field
|
||||
from enum import StrEnum
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, TypedDict
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, TypedDict, cast
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import tomli_w
|
||||
|
||||
from deepagents_cli import auth_store
|
||||
from deepagents_cli import _env_vars, auth_store
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Mapping
|
||||
@@ -487,6 +487,17 @@ OPTIONAL_AUTH_ENV: dict[str, str] = {"ollama": "OLLAMA_API_KEY"}
|
||||
PROVIDER_HOST_ENV: dict[str, str] = {"ollama": "OLLAMA_HOST"}
|
||||
"""Provider-specific env vars that can point a local provider at a remote host."""
|
||||
|
||||
OLLAMA_DEFAULT_BASE_URL = "http://localhost:11434"
|
||||
"""Default endpoint assumed when no `base_url` or `OLLAMA_HOST` is configured."""
|
||||
|
||||
OLLAMA_DISCOVERY_TIMEOUT_SECONDS = 1.0
|
||||
"""Socket timeout for Ollama discovery probes.
|
||||
|
||||
Kept short so a dead daemon does not stall switcher loading. Discovery runs
|
||||
off the UI loop in a worker thread and may call `/api/tags` and `/api/show`,
|
||||
so this caps the worst-case wait visible to the user.
|
||||
"""
|
||||
|
||||
|
||||
# Module-level caches — cleared by `clear_caches()`.
|
||||
_available_models_cache: dict[str, list[str]] | None = None
|
||||
@@ -494,6 +505,8 @@ _builtin_providers_cache: dict[str, Any] | None = None
|
||||
_default_config_cache: ModelConfig | None = None
|
||||
_provider_profiles_cache: dict[str, dict[str, Any]] = {}
|
||||
_provider_profiles_lock = threading.Lock()
|
||||
_ollama_installed_models_cache: dict[str, list[str]] = {}
|
||||
_ollama_model_profiles_cache: dict[tuple[str, str], dict[str, Any]] = {}
|
||||
_profiles_cache: Mapping[str, ModelProfileEntry] | None = None
|
||||
_profiles_override_cache: tuple[int, Mapping[str, ModelProfileEntry]] | None = None
|
||||
|
||||
@@ -508,6 +521,8 @@ def clear_caches() -> None:
|
||||
_builtin_providers_cache = None
|
||||
_default_config_cache = None
|
||||
_provider_profiles_cache.clear()
|
||||
_ollama_installed_models_cache.clear()
|
||||
_ollama_model_profiles_cache.clear()
|
||||
_profiles_cache = None
|
||||
_profiles_override_cache = None
|
||||
invalidate_thread_config_cache()
|
||||
@@ -771,6 +786,31 @@ def get_available_models() -> dict[str, list[str]]:
|
||||
if model not in existing:
|
||||
available[provider_name].append(model)
|
||||
|
||||
# `langchain-ollama` ships no profile data, so the steps above leave the
|
||||
# switcher empty unless the user hand-curates `models = [...]` in config.
|
||||
# Probe the daemon for installed models and merge them in,
|
||||
# preserving explicit config order (config wins) with discoveries appended.
|
||||
# Cached alongside the rest of `available`; refresh by
|
||||
# calling `clear_caches()` (e.g. via the `/reload` slash command).
|
||||
if (
|
||||
_ollama_discovery_enabled()
|
||||
and "ollama" in registry_providers
|
||||
and config.is_provider_enabled("ollama")
|
||||
and importlib.util.find_spec("langchain_ollama") is not None
|
||||
):
|
||||
endpoint = _get_provider_endpoint("ollama", config)
|
||||
discovered = _get_ollama_installed_models(endpoint)
|
||||
if discovered:
|
||||
available["ollama"] = list(
|
||||
dict.fromkeys([*available.get("ollama", []), *discovered])
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
"Ollama discovery returned no models for %s; "
|
||||
"daemon may be down or have no pulls",
|
||||
endpoint or OLLAMA_DEFAULT_BASE_URL,
|
||||
)
|
||||
|
||||
_available_models_cache = available
|
||||
return available
|
||||
|
||||
@@ -928,6 +968,42 @@ def get_model_profiles(
|
||||
)
|
||||
result[spec] = _build_entry({}, overrides, cli_override)
|
||||
|
||||
# `langchain-ollama` does not ship static profile data. When discovery is
|
||||
# enabled, ask the daemon for model metadata so the selector can show
|
||||
# context length and capabilities for locally pulled models.
|
||||
if (
|
||||
_ollama_discovery_enabled()
|
||||
and "ollama" in registry_providers
|
||||
and config.is_provider_enabled("ollama")
|
||||
and importlib.util.find_spec("langchain_ollama") is not None
|
||||
):
|
||||
endpoint = _get_provider_endpoint("ollama", config)
|
||||
discovered_model_names = _get_ollama_installed_models(endpoint)
|
||||
configured_model_names = [
|
||||
spec.removeprefix("ollama:")
|
||||
for spec in result
|
||||
if spec.startswith("ollama:")
|
||||
]
|
||||
model_names = list(
|
||||
dict.fromkeys([*configured_model_names, *discovered_model_names])
|
||||
)
|
||||
if model_names:
|
||||
discovered_profiles = _fetch_ollama_installed_model_profiles(
|
||||
endpoint,
|
||||
model_names,
|
||||
)
|
||||
for model_name in model_names:
|
||||
profile = discovered_profiles.get(model_name, {})
|
||||
spec = f"ollama:{model_name}"
|
||||
existing = result.get(spec)
|
||||
base = dict(existing["profile"]) if existing is not None else {}
|
||||
base.update(profile)
|
||||
overrides = config.get_profile_overrides(
|
||||
"ollama", model_name=model_name
|
||||
)
|
||||
result[spec] = _build_entry(base, overrides, cli_override)
|
||||
seen_specs.add(spec)
|
||||
|
||||
frozen = MappingProxyType(result)
|
||||
if cli_override is None:
|
||||
_profiles_cache = frozen
|
||||
@@ -978,6 +1054,309 @@ def _get_provider_endpoint(provider: str, config: ModelConfig) -> str | None:
|
||||
return resolve_env_var(host_env)
|
||||
|
||||
|
||||
_OLLAMA_DISCOVERY_FALSY: frozenset[str] = frozenset({"0", "false", "no", "off"})
|
||||
"""Normalized values that disable Ollama discovery when set in `OLLAMA_DISCOVERY`."""
|
||||
|
||||
_OLLAMA_DISCOVERY_TRUTHY: frozenset[str] = frozenset({"1", "true", "yes", "on"})
|
||||
"""Normalized values that enable Ollama discovery when set in `OLLAMA_DISCOVERY`."""
|
||||
|
||||
|
||||
def _ollama_discovery_enabled() -> bool:
|
||||
"""Return whether Ollama model/profile discovery may run.
|
||||
|
||||
Defaults to enabled. Opt out via `_env_vars.OLLAMA_DISCOVERY` set to a
|
||||
falsy value (`0`, `false`, `no`, `off`); truthy values (`1`, `true`,
|
||||
`yes`, `on`) explicitly enable. Unrecognized values warn and fall through
|
||||
to the default because the user clearly tried to configure something.
|
||||
"""
|
||||
raw = resolve_env_var(_env_vars.OLLAMA_DISCOVERY)
|
||||
if raw is None:
|
||||
return True
|
||||
normalized = raw.strip().lower()
|
||||
if normalized in _OLLAMA_DISCOVERY_FALSY:
|
||||
return False
|
||||
if normalized in _OLLAMA_DISCOVERY_TRUTHY:
|
||||
return True
|
||||
logger.warning(
|
||||
"Unrecognized value for %s: %r; expected one of %s. Defaulting to enabled.",
|
||||
_env_vars.OLLAMA_DISCOVERY,
|
||||
raw,
|
||||
sorted(_OLLAMA_DISCOVERY_FALSY | _OLLAMA_DISCOVERY_TRUTHY),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _get_ollama_installed_models(endpoint: str | None) -> list[str]:
|
||||
"""Return cached Ollama model names for `endpoint`.
|
||||
|
||||
Args:
|
||||
endpoint: Base URL of the Ollama daemon. When `None`, defaults to
|
||||
`OLLAMA_DEFAULT_BASE_URL`.
|
||||
|
||||
Returns:
|
||||
Sorted list of model names reported by `/api/tags`.
|
||||
"""
|
||||
key = (endpoint or OLLAMA_DEFAULT_BASE_URL).rstrip("/")
|
||||
cached = _ollama_installed_models_cache.get(key)
|
||||
if cached is not None:
|
||||
return list(cached)
|
||||
models = _fetch_ollama_installed_models(endpoint)
|
||||
if models:
|
||||
_ollama_installed_models_cache[key] = models
|
||||
return list(models)
|
||||
|
||||
|
||||
def _fetch_ollama_installed_models(
|
||||
endpoint: str | None,
|
||||
*,
|
||||
timeout: float = OLLAMA_DISCOVERY_TIMEOUT_SECONDS,
|
||||
) -> list[str]:
|
||||
"""Discover models installed in a local or hosted Ollama daemon.
|
||||
|
||||
Issues a `GET {endpoint}/api/tags` and returns the sorted list of model
|
||||
names reported by the daemon. The probe is best-effort: any error
|
||||
(timeout, connection refused, malformed JSON) yields an empty list and is
|
||||
logged at debug level so the model switcher can fall back gracefully.
|
||||
|
||||
When probing a local endpoint and `OLLAMA_API_KEY` (or the
|
||||
`DEEPAGENTS_CLI_`-prefixed variant) is set, its value is forwarded as a
|
||||
`Bearer` token. Discovery never forwards credentials to non-local endpoints.
|
||||
|
||||
Args:
|
||||
endpoint: Base URL of the Ollama daemon. When `None`, defaults to
|
||||
`OLLAMA_DEFAULT_BASE_URL`. A trailing `/` is tolerated.
|
||||
timeout: Socket timeout in seconds.
|
||||
|
||||
Returns:
|
||||
Sorted list of model names; empty when the daemon is unreachable or
|
||||
returns no models.
|
||||
"""
|
||||
import json
|
||||
from urllib.error import URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
base = (endpoint or OLLAMA_DEFAULT_BASE_URL).rstrip("/")
|
||||
if not base.startswith(("http://", "https://")):
|
||||
logger.warning(
|
||||
"Skipping Ollama discovery: %r has no http:// or https:// scheme. "
|
||||
"Set base_url or OLLAMA_HOST to e.g. http://localhost:11434.",
|
||||
base,
|
||||
)
|
||||
return []
|
||||
url = f"{base}/api/tags"
|
||||
|
||||
headers = _ollama_discovery_headers(base, content_type=False)
|
||||
request = Request(url, headers=headers) # noqa: S310 # scheme guarded above
|
||||
# Catch-all is intentional: discovery is best-effort and must never break
|
||||
# the model selector. The narrow tuple is fully subsumed by `Exception`
|
||||
# below; we keep it only to log expected transport failures at debug while
|
||||
# surfacing unexpected ones at warning so a real bug doesn't disappear.
|
||||
# Notably catches `pytest-socket`'s `SocketBlockedError`, which inherits
|
||||
# from `Exception` (not `OSError`) and would otherwise propagate during
|
||||
# unit tests run with `--disable-socket`. `KeyboardInterrupt` and
|
||||
# `SystemExit` derive from `BaseException` and bypass both branches.
|
||||
try:
|
||||
with urlopen(request, timeout=timeout) as response: # noqa: S310 # scheme guarded above
|
||||
payload = json.loads(response.read().decode("utf-8"))
|
||||
except (URLError, TimeoutError, OSError, ValueError) as exc:
|
||||
logger.debug("Ollama model discovery failed for %s: %s", url, exc)
|
||||
return []
|
||||
except Exception as exc: # noqa: BLE001 # see comment above
|
||||
logger.warning(
|
||||
"Ollama model discovery raised unexpected %s for %s: %s",
|
||||
type(exc).__name__,
|
||||
url,
|
||||
exc,
|
||||
)
|
||||
return []
|
||||
|
||||
if not isinstance(payload, dict) or not isinstance(payload.get("models"), list):
|
||||
logger.debug(
|
||||
"Ollama discovery: %s returned unexpected payload shape (%s); "
|
||||
"endpoint may not be an Ollama daemon",
|
||||
url,
|
||||
type(payload).__name__,
|
||||
)
|
||||
return []
|
||||
|
||||
names: list[str] = []
|
||||
for entry in payload["models"]:
|
||||
if isinstance(entry, dict):
|
||||
name = entry.get("name")
|
||||
if isinstance(name, str) and name:
|
||||
names.append(name)
|
||||
names.sort()
|
||||
return names
|
||||
|
||||
|
||||
def _ollama_discovery_headers(endpoint: str, *, content_type: bool) -> dict[str, str]:
|
||||
"""Build headers for Ollama discovery requests.
|
||||
|
||||
Args:
|
||||
endpoint: Base URL for the discovery request.
|
||||
content_type: Whether to include a JSON `Content-Type` header.
|
||||
|
||||
Returns:
|
||||
HTTP headers including optional bearer auth for local endpoints.
|
||||
"""
|
||||
headers: dict[str, str] = {"Accept": "application/json"}
|
||||
if content_type:
|
||||
headers["Content-Type"] = "application/json"
|
||||
optional_env = OPTIONAL_AUTH_ENV.get("ollama")
|
||||
if optional_env and _is_local_endpoint(endpoint):
|
||||
api_key = resolve_env_var(optional_env)
|
||||
if api_key:
|
||||
headers["Authorization"] = f"Bearer {api_key}"
|
||||
return headers
|
||||
|
||||
|
||||
def _coerce_positive_int(value: object) -> int | None:
|
||||
"""Return `value` as a positive integer, or `None` when unavailable."""
|
||||
if isinstance(value, bool):
|
||||
return None
|
||||
if isinstance(value, int) and value > 0:
|
||||
return value
|
||||
if isinstance(value, float) and value > 0 and value.is_integer():
|
||||
return int(value)
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
parsed = int(value)
|
||||
except ValueError:
|
||||
return None
|
||||
if parsed > 0:
|
||||
return parsed
|
||||
return None
|
||||
|
||||
|
||||
def _profile_from_ollama_show_payload(payload: object) -> dict[str, Any]:
|
||||
"""Extract LangChain-style profile fields from an Ollama `/api/show` payload.
|
||||
|
||||
Args:
|
||||
payload: Decoded JSON response from `POST /api/show`.
|
||||
|
||||
Returns:
|
||||
Profile fields understood by the model selector, such as
|
||||
`max_input_tokens` and `tool_calling`.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
return {}
|
||||
payload_dict = cast("dict[str, object]", payload)
|
||||
|
||||
profile: dict[str, Any] = {}
|
||||
model_info = payload_dict.get("model_info")
|
||||
if isinstance(model_info, dict):
|
||||
context_lengths = [
|
||||
length
|
||||
for key, value in model_info.items()
|
||||
if isinstance(key, str)
|
||||
and (key == "context_length" or key.endswith(".context_length"))
|
||||
and (length := _coerce_positive_int(value)) is not None
|
||||
]
|
||||
if context_lengths:
|
||||
profile["max_input_tokens"] = max(context_lengths)
|
||||
|
||||
capabilities = payload_dict.get("capabilities")
|
||||
if isinstance(capabilities, list):
|
||||
capability_names = {item for item in capabilities if isinstance(item, str)}
|
||||
if "completion" in capability_names:
|
||||
profile["text_inputs"] = True
|
||||
profile["text_outputs"] = True
|
||||
if "tools" in capability_names:
|
||||
profile["tool_calling"] = True
|
||||
if "thinking" in capability_names:
|
||||
profile["reasoning_output"] = True
|
||||
|
||||
if not profile and ("model_info" in payload_dict or "capabilities" in payload_dict):
|
||||
logger.debug(
|
||||
"Ollama profile discovery returned a payload with no recognized "
|
||||
"profile fields; top-level keys: %s",
|
||||
sorted(str(key) for key in payload_dict),
|
||||
)
|
||||
|
||||
return profile
|
||||
|
||||
|
||||
def _fetch_ollama_installed_model_profiles(
|
||||
endpoint: str | None,
|
||||
model_names: list[str],
|
||||
*,
|
||||
timeout: float = OLLAMA_DISCOVERY_TIMEOUT_SECONDS,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Discover profile metadata for installed Ollama models.
|
||||
|
||||
Issues `POST {endpoint}/api/show` for each model. The probe is best-effort:
|
||||
failures for one model are logged and do not stop profile discovery for the
|
||||
remaining models.
|
||||
|
||||
Args:
|
||||
endpoint: Base URL of the Ollama daemon. When `None`, defaults to
|
||||
`OLLAMA_DEFAULT_BASE_URL`. A trailing `/` is tolerated.
|
||||
model_names: Model names to inspect.
|
||||
timeout: Socket timeout in seconds.
|
||||
|
||||
Returns:
|
||||
Mapping of model name to extracted profile fields.
|
||||
"""
|
||||
import json
|
||||
from urllib.error import URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
base = (endpoint or OLLAMA_DEFAULT_BASE_URL).rstrip("/")
|
||||
if not base.startswith(("http://", "https://")):
|
||||
logger.warning(
|
||||
"Skipping Ollama profile discovery: %r has no http:// or https:// scheme. "
|
||||
"Set base_url or OLLAMA_HOST to e.g. http://localhost:11434.",
|
||||
base,
|
||||
)
|
||||
return {}
|
||||
|
||||
url = f"{base}/api/show"
|
||||
profiles: dict[str, dict[str, Any]] = {}
|
||||
headers = _ollama_discovery_headers(base, content_type=True)
|
||||
|
||||
for model_name in model_names:
|
||||
cache_key = (base, model_name)
|
||||
cached = _ollama_model_profiles_cache.get(cache_key)
|
||||
if cached is not None:
|
||||
profiles[model_name] = dict(cached)
|
||||
continue
|
||||
|
||||
body = json.dumps({"model": model_name}).encode("utf-8")
|
||||
request = Request( # noqa: S310 # scheme guarded above
|
||||
url,
|
||||
data=body,
|
||||
headers=headers,
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urlopen(request, timeout=timeout) as response: # noqa: S310 # scheme guarded above
|
||||
payload = json.loads(response.read().decode("utf-8"))
|
||||
except (URLError, TimeoutError, OSError, ValueError) as exc:
|
||||
logger.debug(
|
||||
"Ollama profile discovery failed for %s via %s: %s",
|
||||
model_name,
|
||||
url,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
except Exception as exc: # noqa: BLE001 # see _fetch_ollama_installed_models
|
||||
logger.warning(
|
||||
"Ollama profile discovery raised unexpected %s for %s via %s: %s",
|
||||
type(exc).__name__,
|
||||
model_name,
|
||||
url,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
|
||||
profile = _profile_from_ollama_show_payload(payload)
|
||||
if profile:
|
||||
_ollama_model_profiles_cache[cache_key] = profile
|
||||
profiles[model_name] = profile
|
||||
|
||||
return profiles
|
||||
|
||||
|
||||
def _has_stored_credential(provider: str) -> bool:
|
||||
"""Return whether `provider` has a credential persisted via `/auth`.
|
||||
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
"""Tests for model_config module."""
|
||||
|
||||
import io
|
||||
import logging
|
||||
from collections.abc import Iterator
|
||||
from contextlib import AbstractContextManager
|
||||
from pathlib import Path
|
||||
from typing import Any, ClassVar
|
||||
from typing import Any, ClassVar, cast
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
@@ -1311,6 +1313,734 @@ api_key_env = "SOME_KEY"
|
||||
assert "empty" not in models
|
||||
|
||||
|
||||
class TestOllamaModelDiscovery:
|
||||
"""Tests for auto-populating the switcher from a running Ollama daemon."""
|
||||
|
||||
@staticmethod
|
||||
def _patch_registry() -> AbstractContextManager[object]:
|
||||
"""Patch the langchain registry so `ollama` is a known provider."""
|
||||
return patch(
|
||||
"deepagents_cli.model_config._get_builtin_providers",
|
||||
return_value={
|
||||
"ollama": ("langchain_ollama.chat_models", "ChatOllama"),
|
||||
},
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _empty_profiles_loader(module_path: str) -> dict[str, Any]:
|
||||
"""Pretend `langchain_ollama` ships no profile data."""
|
||||
if module_path == "langchain_ollama.data._profiles":
|
||||
return {}
|
||||
msg = "not installed"
|
||||
raise ImportError(msg)
|
||||
|
||||
def test_discovery_merges_models_into_switcher(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Daemon-reported models populate `available["ollama"]`."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=["llama3", "qwen3:4b"],
|
||||
),
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
models = get_available_models()
|
||||
|
||||
assert models.get("ollama") == ["llama3", "qwen3:4b"]
|
||||
|
||||
def test_discovery_unions_with_explicit_config_models(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Explicit `models = […]` config still wins / supplements discovery."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("""
|
||||
[models.providers.ollama]
|
||||
models = ["my-finetune"]
|
||||
""")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=["llama3", "my-finetune"],
|
||||
),
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
models = get_available_models()
|
||||
|
||||
# Explicit config first, then newly discovered names; no duplicates.
|
||||
assert models["ollama"] == ["my-finetune", "llama3"]
|
||||
|
||||
def test_discovery_skipped_when_package_missing(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""No HTTP probe when `langchain-ollama` is not installed."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
) as fetch,
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
models = get_available_models()
|
||||
|
||||
fetch.assert_not_called()
|
||||
assert "ollama" not in models
|
||||
|
||||
def test_discovery_disabled_via_env_var(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""`DEEPAGENTS_CLI_OLLAMA_DISCOVERY=0` opts out of the probe."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("")
|
||||
monkeypatch.setenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", "0")
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
) as fetch,
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
models = get_available_models()
|
||||
|
||||
fetch.assert_not_called()
|
||||
assert "ollama" not in models
|
||||
|
||||
def test_discovery_skipped_when_provider_disabled(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""`enabled = false` for ollama prevents the probe."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("""
|
||||
[models.providers.ollama]
|
||||
enabled = false
|
||||
""")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
) as fetch,
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
models = get_available_models()
|
||||
|
||||
fetch.assert_not_called()
|
||||
assert "ollama" not in models
|
||||
|
||||
def test_discovery_warns_on_unknown_env_value(
|
||||
self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Unrecognized env values warn and keep discovery enabled."""
|
||||
monkeypatch.setenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", "maybe")
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
assert model_config._ollama_discovery_enabled() is True
|
||||
|
||||
assert "Unrecognized value for DEEPAGENTS_CLI_OLLAMA_DISCOVERY" in caplog.text
|
||||
|
||||
def test_installed_model_discovery_cached_across_profile_load(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Available-model and profile loading share one `/api/tags` probe."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=["qwen3:4b"],
|
||||
) as fetch,
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_model_profiles",
|
||||
return_value={},
|
||||
),
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
get_available_models()
|
||||
get_model_profiles()
|
||||
|
||||
fetch.assert_called_once_with(None)
|
||||
|
||||
def test_empty_installed_model_discovery_not_cached(self) -> None:
|
||||
"""Empty `/api/tags` results do not block later recovery."""
|
||||
with patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
side_effect=[[], ["qwen3:4b"]],
|
||||
) as fetch:
|
||||
assert model_config._get_ollama_installed_models(None) == []
|
||||
assert model_config._get_ollama_installed_models(None) == ["qwen3:4b"]
|
||||
|
||||
assert fetch.call_count == 2
|
||||
|
||||
def test_model_profiles_include_discovered_context_length(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Discovered Ollama metadata populates model profile entries."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=["qwen3:4b"],
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_model_profiles",
|
||||
return_value={
|
||||
"qwen3:4b": {
|
||||
"max_input_tokens": 262144,
|
||||
"text_inputs": True,
|
||||
"text_outputs": True,
|
||||
"tool_calling": True,
|
||||
"reasoning_output": True,
|
||||
},
|
||||
},
|
||||
),
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
profiles = get_model_profiles()
|
||||
|
||||
entry = profiles["ollama:qwen3:4b"]
|
||||
assert entry["profile"]["max_input_tokens"] == 262144
|
||||
assert entry["profile"]["tool_calling"] is True
|
||||
assert entry["profile"]["reasoning_output"] is True
|
||||
assert entry["overridden_keys"] == frozenset()
|
||||
|
||||
def test_model_profiles_apply_config_overrides_to_discovered_metadata(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Config profile values still override Ollama-discovered metadata."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("""
|
||||
[models.providers.ollama.profile."qwen3:4b"]
|
||||
max_input_tokens = 4096
|
||||
""")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=["qwen3:4b"],
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_model_profiles",
|
||||
return_value={"qwen3:4b": {"max_input_tokens": 262144}},
|
||||
),
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
profiles = get_model_profiles()
|
||||
|
||||
entry = profiles["ollama:qwen3:4b"]
|
||||
assert entry["profile"]["max_input_tokens"] == 4096
|
||||
assert "max_input_tokens" in entry["overridden_keys"]
|
||||
|
||||
def test_model_profiles_fetch_configured_models_when_tags_empty(
|
||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Configured Ollama models are inspected even when `/api/tags` is empty."""
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text("""
|
||||
[models.providers.ollama]
|
||||
models = ["qwen3:4b"]
|
||||
""")
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_DISCOVERY", raising=False)
|
||||
|
||||
with (
|
||||
self._patch_registry(),
|
||||
patch(
|
||||
"deepagents_cli.model_config._load_provider_profiles",
|
||||
side_effect=self._empty_profiles_loader,
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config.importlib.util.find_spec",
|
||||
return_value=object(),
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_models",
|
||||
return_value=[],
|
||||
),
|
||||
patch(
|
||||
"deepagents_cli.model_config._fetch_ollama_installed_model_profiles",
|
||||
return_value={"qwen3:4b": {"max_input_tokens": 262144}},
|
||||
) as fetch_profiles,
|
||||
patch.object(model_config, "DEFAULT_CONFIG_PATH", config_path),
|
||||
):
|
||||
profiles = get_model_profiles()
|
||||
|
||||
fetch_profiles.assert_called_once_with(None, ["qwen3:4b"])
|
||||
assert profiles["ollama:qwen3:4b"]["profile"]["max_input_tokens"] == 262144
|
||||
|
||||
|
||||
class _BytesContext:
|
||||
"""Minimal context manager wrapping a bytes payload for fake `urlopen`."""
|
||||
|
||||
def __init__(self, body: bytes) -> None:
|
||||
self._body = io.BytesIO(body)
|
||||
|
||||
def __enter__(self) -> io.BytesIO:
|
||||
return self._body
|
||||
|
||||
def __exit__(self, *_exc: object) -> None:
|
||||
self._body.close()
|
||||
|
||||
|
||||
class TestFetchOllamaInstalledModels:
|
||||
"""Tests for the `_fetch_ollama_installed_models` HTTP probe."""
|
||||
|
||||
def test_returns_sorted_names_from_payload(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Parses `{"models": [{"name": ...}]}` and sorts results."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_url: list[str] = []
|
||||
captured_timeout: list[float] = []
|
||||
captured_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_urlopen(request: Request, timeout: float) -> _BytesContext:
|
||||
captured_url.append(request.full_url)
|
||||
captured_timeout.append(timeout)
|
||||
captured_headers.append(dict(request.header_items()))
|
||||
payload = {"models": [{"name": "qwen3:4b"}, {"name": "llama3"}]}
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
result = model_config._fetch_ollama_installed_models(
|
||||
"http://localhost:11434"
|
||||
)
|
||||
|
||||
assert result == ["llama3", "qwen3:4b"]
|
||||
assert captured_url == ["http://localhost:11434/api/tags"]
|
||||
assert captured_timeout == [model_config.OLLAMA_DISCOVERY_TIMEOUT_SECONDS]
|
||||
assert "Authorization" not in {k.title() for k in captured_headers[0]}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
{},
|
||||
{"models": "qwen3:4b"},
|
||||
],
|
||||
)
|
||||
def test_returns_empty_for_unexpected_payload_shape(
|
||||
self, payload: dict[str, object], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Missing or non-list `models` payloads are ignored."""
|
||||
import json
|
||||
|
||||
def fake_urlopen(*_args: object, **_kwargs: object) -> _BytesContext:
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
result = model_config._fetch_ollama_installed_models(
|
||||
"http://localhost:11434"
|
||||
)
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_returns_empty_for_malformed_json(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Malformed JSON is treated as discovery failure."""
|
||||
|
||||
def fake_urlopen(*_args: object, **_kwargs: object) -> _BytesContext:
|
||||
return _BytesContext(b"{not json")
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
result = model_config._fetch_ollama_installed_models(
|
||||
"http://localhost:11434"
|
||||
)
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_silent_on_connection_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Connection errors yield an empty list without raising."""
|
||||
from urllib.error import URLError
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
url_error = URLError("connection refused")
|
||||
|
||||
def boom(*_args: object, **_kwargs: object) -> None:
|
||||
raise url_error
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=boom):
|
||||
assert model_config._fetch_ollama_installed_models(None) == []
|
||||
|
||||
def test_uses_default_endpoint_when_none(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Falls back to `OLLAMA_DEFAULT_BASE_URL` when no endpoint is given."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_url: list[str] = []
|
||||
|
||||
def fake_urlopen(
|
||||
request: Request,
|
||||
timeout: float, # noqa: ARG001
|
||||
) -> _BytesContext:
|
||||
captured_url.append(request.full_url)
|
||||
return _BytesContext(json.dumps({"models": []}).encode("utf-8"))
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
assert model_config._fetch_ollama_installed_models(None) == []
|
||||
|
||||
assert captured_url[0].startswith(model_config.OLLAMA_DEFAULT_BASE_URL)
|
||||
assert captured_url[0].endswith("/api/tags")
|
||||
|
||||
def test_forwards_optional_api_key_header(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""`OLLAMA_API_KEY` is forwarded to local discovery endpoints."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_urlopen(
|
||||
request: Request,
|
||||
timeout: float, # noqa: ARG001
|
||||
) -> _BytesContext:
|
||||
captured_headers.append(dict(request.header_items()))
|
||||
return _BytesContext(json.dumps({"models": []}).encode("utf-8"))
|
||||
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "secret-token")
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
model_config._fetch_ollama_installed_models("http://localhost:11434")
|
||||
|
||||
# Header names are title-cased by urllib.
|
||||
assert captured_headers[0].get("Authorization") == "Bearer secret-token"
|
||||
|
||||
def test_does_not_forward_optional_api_key_to_remote_endpoint(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Discovery does not send credentials to non-local endpoints."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_urlopen(
|
||||
request: Request,
|
||||
timeout: float, # noqa: ARG001
|
||||
) -> _BytesContext:
|
||||
captured_headers.append(dict(request.header_items()))
|
||||
return _BytesContext(json.dumps({"models": []}).encode("utf-8"))
|
||||
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "secret-token")
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
model_config._fetch_ollama_installed_models("https://ollama.example.com")
|
||||
|
||||
assert "Authorization" not in captured_headers[0]
|
||||
|
||||
def test_rejects_unsupported_scheme(self) -> None:
|
||||
"""Non-http(s) endpoints are skipped without invoking the network."""
|
||||
with patch("urllib.request.urlopen") as fake:
|
||||
assert (
|
||||
model_config._fetch_ollama_installed_models("ftp://localhost:11434")
|
||||
== []
|
||||
)
|
||||
fake.assert_not_called()
|
||||
|
||||
|
||||
class TestFetchOllamaInstalledModelProfiles:
|
||||
"""Tests for Ollama `/api/show` profile discovery."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
(True, None),
|
||||
(False, None),
|
||||
(-1, None),
|
||||
(0, None),
|
||||
(1.5, None),
|
||||
("4096", 4096),
|
||||
(None, None),
|
||||
],
|
||||
)
|
||||
def test_coerce_positive_int_edges(
|
||||
self, value: object, expected: int | None
|
||||
) -> None:
|
||||
"""Only positive whole-number values are accepted."""
|
||||
assert model_config._coerce_positive_int(value) == expected
|
||||
|
||||
def test_extracts_profile_from_show_payload(self) -> None:
|
||||
"""Context length and capabilities become selector profile fields."""
|
||||
payload = {
|
||||
"model_info": {
|
||||
"general.architecture": "qwen3",
|
||||
"qwen3.context_length": 262144,
|
||||
"qwen3.embedding_length": 2560,
|
||||
},
|
||||
"capabilities": ["completion", "tools", "thinking"],
|
||||
}
|
||||
|
||||
profile = model_config._profile_from_ollama_show_payload(payload)
|
||||
|
||||
assert profile == {
|
||||
"max_input_tokens": 262144,
|
||||
"text_inputs": True,
|
||||
"text_outputs": True,
|
||||
"tool_calling": True,
|
||||
"reasoning_output": True,
|
||||
}
|
||||
|
||||
def test_extracts_max_from_multiple_context_lengths(self) -> None:
|
||||
"""When several context lengths are present, the largest is used."""
|
||||
payload = {
|
||||
"model_info": {
|
||||
"context_length": 8192,
|
||||
"draft.context_length": 4096,
|
||||
"qwen3.context_length": 262144,
|
||||
},
|
||||
}
|
||||
|
||||
profile = model_config._profile_from_ollama_show_payload(payload)
|
||||
|
||||
assert profile == {"max_input_tokens": 262144}
|
||||
|
||||
def test_non_dict_payload_returns_empty_profile(self) -> None:
|
||||
"""Malformed payloads are ignored."""
|
||||
assert model_config._profile_from_ollama_show_payload([]) == {}
|
||||
|
||||
def test_missing_model_info_returns_capabilities_only(self) -> None:
|
||||
"""Capabilities can still be extracted without model metadata."""
|
||||
payload = {"capabilities": ["tools"]}
|
||||
|
||||
profile = model_config._profile_from_ollama_show_payload(payload)
|
||||
|
||||
assert profile == {"tool_calling": True}
|
||||
|
||||
def test_non_list_capabilities_ignored(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Unexpected capability shape does not produce false flags."""
|
||||
payload = {"model_info": {}, "capabilities": "tools"}
|
||||
|
||||
with caplog.at_level(logging.DEBUG):
|
||||
profile = model_config._profile_from_ollama_show_payload(payload)
|
||||
|
||||
assert profile == {}
|
||||
assert "no recognized profile fields" in caplog.text
|
||||
|
||||
def test_posts_model_names_to_show_endpoint(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Fetches local `/api/show` with bearer auth and parses context length."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_url: list[str] = []
|
||||
captured_body: list[dict[str, str]] = []
|
||||
captured_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_urlopen(request: Request, timeout: float) -> _BytesContext:
|
||||
assert timeout == model_config.OLLAMA_DISCOVERY_TIMEOUT_SECONDS
|
||||
captured_url.append(request.full_url)
|
||||
captured_headers.append(dict(request.header_items()))
|
||||
data = cast("bytes", request.data)
|
||||
captured_body.append(json.loads(data.decode("utf-8")))
|
||||
payload = {
|
||||
"model_info": {"qwen3.context_length": 262144},
|
||||
"capabilities": ["completion", "tools"],
|
||||
}
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "secret-token")
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
profiles = model_config._fetch_ollama_installed_model_profiles(
|
||||
"http://localhost:11434",
|
||||
["qwen3:4b"],
|
||||
)
|
||||
|
||||
assert profiles["qwen3:4b"]["max_input_tokens"] == 262144
|
||||
assert profiles["qwen3:4b"]["tool_calling"] is True
|
||||
assert captured_url == ["http://localhost:11434/api/show"]
|
||||
assert captured_body == [{"model": "qwen3:4b"}]
|
||||
assert captured_headers[0].get("Authorization") == "Bearer secret-token"
|
||||
|
||||
def test_show_does_not_forward_optional_api_key_to_remote_endpoint(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Profile discovery does not send credentials to non-local endpoints."""
|
||||
import json
|
||||
from urllib.request import Request
|
||||
|
||||
captured_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_urlopen(
|
||||
request: Request,
|
||||
timeout: float, # noqa: ARG001
|
||||
) -> _BytesContext:
|
||||
captured_headers.append(dict(request.header_items()))
|
||||
payload = {"model_info": {"qwen3.context_length": 262144}}
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "secret-token")
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
profiles = model_config._fetch_ollama_installed_model_profiles(
|
||||
"https://ollama.example.com",
|
||||
["qwen3:4b"],
|
||||
)
|
||||
|
||||
assert profiles["qwen3:4b"]["max_input_tokens"] == 262144
|
||||
assert "Authorization" not in captured_headers[0]
|
||||
|
||||
def test_successful_profiles_are_cached(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Repeated profile discovery reuses successful `/api/show` results."""
|
||||
import json
|
||||
|
||||
calls = 0
|
||||
|
||||
def fake_urlopen(*_args: object, **_kwargs: object) -> _BytesContext:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
payload = {"model_info": {"qwen3.context_length": 262144}}
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
|
||||
monkeypatch.delenv("DEEPAGENTS_CLI_OLLAMA_API_KEY", raising=False)
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
first = model_config._fetch_ollama_installed_model_profiles(
|
||||
"http://localhost:11434",
|
||||
["qwen3:4b"],
|
||||
)
|
||||
second = model_config._fetch_ollama_installed_model_profiles(
|
||||
"http://localhost:11434",
|
||||
["qwen3:4b"],
|
||||
)
|
||||
|
||||
assert first == second == {"qwen3:4b": {"max_input_tokens": 262144}}
|
||||
assert calls == 1
|
||||
|
||||
def test_continues_after_per_model_failure(self) -> None:
|
||||
"""A failed model profile lookup does not abort the whole batch."""
|
||||
import json
|
||||
from urllib.error import URLError
|
||||
from urllib.request import Request
|
||||
|
||||
def fake_urlopen(request: Request, timeout: float) -> _BytesContext: # noqa: ARG001
|
||||
data = cast("bytes", request.data)
|
||||
body = json.loads(data.decode("utf-8"))
|
||||
if body["model"] == "broken":
|
||||
msg = "not found"
|
||||
raise URLError(msg)
|
||||
payload = {"model_info": {"llama.context_length": 8192}}
|
||||
return _BytesContext(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
profiles = model_config._fetch_ollama_installed_model_profiles(
|
||||
"http://localhost:11434",
|
||||
["broken", "llama3"],
|
||||
)
|
||||
|
||||
assert profiles == {"llama3": {"max_input_tokens": 8192}}
|
||||
|
||||
|
||||
class TestDisabledProviders:
|
||||
"""Tests for provider hiding via `enabled = false`."""
|
||||
|
||||
@@ -3466,8 +4196,16 @@ max_input_tokens = 4096
|
||||
get_model_profiles()
|
||||
|
||||
assert model_config._profiles_cache is not None
|
||||
model_config._ollama_installed_models_cache["http://localhost:11434"] = [
|
||||
"qwen3:4b"
|
||||
]
|
||||
model_config._ollama_model_profiles_cache[
|
||||
"http://localhost:11434", "qwen3:4b"
|
||||
] = {"max_input_tokens": 262144}
|
||||
clear_caches()
|
||||
assert model_config._profiles_cache is None
|
||||
assert model_config._ollama_installed_models_cache == {}
|
||||
assert model_config._ollama_model_profiles_cache == {}
|
||||
|
||||
def test_overridden_keys_subset_of_profile(self, tmp_path: Path) -> None:
|
||||
"""overridden_keys is always a subset of profile keys."""
|
||||
|
||||
Reference in New Issue
Block a user