llm

zocrates.llm

Central LLM factory, OpenRouter model list fetcher, and connection test helpers for Zocrates.

CACHE_TTL_SECONDS module-attribute

CACHE_TTL_SECONDS = 86400

DEFAULT_LLM_MAX_TOKENS module-attribute

DEFAULT_LLM_MAX_TOKENS = 16384

DEFAULT_LLM_MODEL module-attribute

DEFAULT_LLM_MODEL = 'openrouter/~z-ai/glm-flash-latest'

DEFAULT_LLM_NUM_RETRIES module-attribute

DEFAULT_LLM_NUM_RETRIES = 4

DEFAULT_LLM_TEMPERATURE module-attribute

DEFAULT_LLM_TEMPERATURE = None

DEFAULT_OLLAMA_NUM_CTX module-attribute

DEFAULT_OLLAMA_NUM_CTX = 16384

PROVIDER_DEFAULT_MODELS module-attribute

PROVIDER_DEFAULT_MODELS = {'openrouter_llm': 'openrouter/~z-ai/glm-flash-latest', 'openrouter': 'openrouter/~z-ai/glm-flash-latest', 'ollama': 'llama3.1:8b', 'vllm': 'default', 'openai_compatible': 'default'}

_OPENROUTER_DECISION_CACHE module-attribute

_OPENROUTER_DECISION_CACHE = {'timestamp': 0.0, 'data': []}

_OPENROUTER_EMBED_CACHE module-attribute

_OPENROUTER_EMBED_CACHE = {'timestamp': 0, 'data': []}

_OPENROUTER_LLM_CACHE module-attribute

_OPENROUTER_LLM_CACHE = {'timestamp': 0, 'data': []}

logger module-attribute

logger = logging.getLogger(__name__)

_format_base_url

_format_base_url(base_url, provider)

Format and normalize base URL for local providers.

Source code in zocrates/llm.py

def _format_base_url(base_url: str, provider: str) -> str:
    """Format and normalize base URL for local providers."""
    url = (base_url or "").strip()
    if not url:
        return url
    if not url.lower().startswith(("http://", "https://")):
        url = f"http://{url}"
    else:
        scheme, sep, rest = url.partition("://")
        url = f"{scheme.lower()}{sep}{rest}"
    url = url.rstrip("/")
    if provider == "ollama":
        if url.endswith("/v1"):
            url = url[:-3].rstrip("/")
    elif provider in ("vllm", "openai_compatible"):
        parsed = urlsplit(url)
        if not parsed.path or parsed.path == "/":
            url = f"{url}/v1"
    return url

_get_cache_file_path

_get_cache_file_path(target_type='llm', zotero_data_dir=None)

Return the disk cache file path for OpenRouter models.

Parameters:

Name	Type	Description	Default
target_type	str	Type of model cache ('llm' or 'embedding').	'llm'
zotero_data_dir	Path | str | None	Optional custom Zotero data directory path.	None

Returns:

Type	Description
Path	Path to the JSON cache file.

Source code in zocrates/llm.py

def _get_cache_file_path(target_type: str = "llm", zotero_data_dir: Path | str | None = None) -> Path:
    """Return the disk cache file path for OpenRouter models.

    Args:
        target_type: Type of model cache ('llm' or 'embedding').
        zotero_data_dir: Optional custom Zotero data directory path.

    Returns:
        Path to the JSON cache file.
    """
    data_dir = get_zocrates_data_dir(zotero_data_dir) / "openrouter_models_cache"
    data_dir.mkdir(parents=True, exist_ok=True)
    if target_type == "embedding":
        filename = "embedding_models.json"
    elif target_type == "decision":
        filename = "decision_models.json"
    else:
        filename = "llm_models.json"
    return data_dir / filename

_resolve_model_name

_resolve_model_name(model_slug, provider)

Resolve a raw model slug into a provider-prefixed LiteLLM model identifier.

Handles provider prefixes and prevents cross-provider slug leakage (e.g. Clash #5).

Source code in zocrates/llm.py

def _resolve_model_name(model_slug: str, provider: str) -> str:
    """Resolve a raw model slug into a provider-prefixed LiteLLM model identifier.

    Handles provider prefixes and prevents cross-provider slug leakage (e.g. Clash #5).
    """
    model_name = model_slug.strip() if model_slug else ""
    if not model_name:
        model_name = PROVIDER_DEFAULT_MODELS.get(provider, DEFAULT_LLM_MODEL)

    if provider == "ollama":
        if model_name.startswith("ollama/"):
            stripped = model_name[len("ollama/"):]
            return f"ollama_chat/{stripped}"
        if model_name.startswith("ollama_chat/"):
            return model_name
        if "/" in model_name:
            remote_prefixes = (
                "openrouter/",
                "google/",
                "anthropic/",
                "meta-llama/",
                "deepseek/",
                "mistralai/",
                "qwen/",
            )
            is_remote_prefix = any(model_name.startswith(p) for p in remote_prefixes)
            is_hf = model_name.startswith("hf.co/") or model_name.startswith("huggingface.co/")
            if is_remote_prefix and not is_hf:
                default_ollama = PROVIDER_DEFAULT_MODELS.get("ollama", "llama3.1:8b")
                logger.warning(
                    "Model '%s' appears to be a remote provider model but provider is ollama; falling back to default '%s'",
                    model_name,
                    default_ollama,
                )
                if default_ollama.startswith("ollama_chat/"):
                    return default_ollama
                if default_ollama.startswith("ollama/"):
                    return f"ollama_chat/{default_ollama[len('ollama/'):]}"
                return f"ollama_chat/{default_ollama}"
            else:
                return f"ollama_chat/{model_name}"
        return f"ollama_chat/{model_name}"

    elif provider in ("vllm", "openai_compatible"):
        if model_name.startswith("openai/"):
            return model_name
        return f"openai/{model_name}"

    elif provider in ("openrouter", "openrouter_llm"):
        if model_name.startswith("openrouter/"):
            return model_name
        return f"openrouter/{model_name}"

    return model_name

_sync_fetch_openrouter_models

_sync_fetch_openrouter_models(key, target_type='llm')

Fetch OpenRouter models catalog synchronously via HTTP GET.

Parameters:

Name	Type	Description	Default
key	str	OpenRouter API key for authentication.	required
target_type	str	Type of model catalog to retrieve ('llm' or 'embedding').	'llm'

Returns:

Type	Description
list[dict]	List of raw model dictionaries from the OpenRouter API response.

Source code in zocrates/llm.py

def _sync_fetch_openrouter_models(key: str, target_type: str = "llm") -> list[dict]:
    """Fetch OpenRouter models catalog synchronously via HTTP GET.

    Args:
        key: OpenRouter API key for authentication.
        target_type: Type of model catalog to retrieve ('llm' or 'embedding').

    Returns:
        List of raw model dictionaries from the OpenRouter API response.
    """
    headers = {
        "Authorization": f"Bearer {key}",
        "HTTP-Referer": "https://github.com/alexplaka/Zocrates",
        "X-OpenRouter-Title": "Zocrates",
    }
    if target_type == "embedding":
        endpoint = "https://openrouter.ai/api/v1/embeddings/models"
    elif target_type == "decision":
        endpoint = "https://openrouter.ai/api/v1/models?output_modalities=decisions"
    else:
        endpoint = "https://openrouter.ai/api/v1/models"
    response = requests.get(endpoint, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json().get("data", [])

build_agent_llm

build_agent_llm(agent_cfg, global_cfg=None, zotero_data_dir=None)

Build a dedicated LiteLlm instance for an agent role.

Resolves configuration according to precedence

agent_cfg > global_cfg > defaults

Parameters:

Name	Type	Description	Default
agent_cfg	dict[str, Any]	Role-specific override dict (e.g. from override_agent_models).	required
global_cfg	dict[str, Any] | None	Optional global configuration dictionary (loaded if None).	None
zotero_data_dir	Path | str | None	Optional Zotero data directory path.	None

Returns:

Type	Description
BaseLlm	Configured ADK BaseLlm (LiteLlm) instance.

Source code in zocrates/llm.py

def build_agent_llm(
    agent_cfg: dict[str, Any],
    global_cfg: dict[str, Any] | None = None,
    zotero_data_dir: Path | str | None = None,
) -> BaseLlm:
    """Build a dedicated LiteLlm instance for an agent role.

    Resolves configuration according to precedence:
      agent_cfg > global_cfg > defaults

    Args:
        agent_cfg: Role-specific override dict (e.g. from override_agent_models).
        global_cfg: Optional global configuration dictionary (loaded if None).
        zotero_data_dir: Optional Zotero data directory path.

    Returns:
        Configured ADK BaseLlm (LiteLlm) instance.
    """
    if global_cfg is None:
        global_cfg = load_zocrates_config(zotero_data_dir)

    raw_provider = agent_cfg.get("provider")
    raw_model = (
        agent_cfg.get("model")
        or global_cfg.get("llm_model")
        or PROVIDER_DEFAULT_MODELS.get(raw_provider or global_cfg.get("llm_provider", "openrouter_llm"), DEFAULT_LLM_MODEL)
    )

    if raw_provider:
        provider = raw_provider
    elif raw_model.startswith("ollama/") or raw_model.startswith("ollama_chat/"):
        provider = "ollama"
    elif raw_model.startswith("openai/"):
        provider = "vllm"
    elif raw_model.startswith("openrouter/"):
        provider = "openrouter_llm"
    else:
        provider = global_cfg.get("llm_provider") or "openrouter_llm"

    model_name = _resolve_model_name(raw_model, provider)

    kwargs: dict[str, Any] = {}

    # 1. OpenRouter (Remote LLM)
    if provider in ("openrouter", "openrouter_llm"):
        key = (
            agent_cfg.get("api_key")
            or get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))
        )
        if key:
            kwargs["api_key"] = key
        kwargs["extra_headers"] = {
            # "HTTP-Referer": "https://github.com/alexplaka/Zocrates",
            "X-OpenRouter-Title": "Zocrates",
        }

    # 2. Local Ollama
    elif provider == "ollama":
        raw_url = (
            agent_cfg.get("base_url")
            or global_cfg.get("llm_base_url")
            or "http://localhost:11434"
        )
        kwargs["api_base"] = _format_base_url(raw_url, "ollama")
        extra_opts = dict(global_cfg.get("llm_extra_options") or {})
        extra_opts.update(agent_cfg.get("extra_options") or {})
        effective_num_ctx = extra_opts.get("num_ctx") or get_ollama_num_ctx(cfg=global_cfg)
        try:
            kwargs["num_ctx"] = int(effective_num_ctx)
        except (TypeError, ValueError):
            kwargs["num_ctx"] = DEFAULT_OLLAMA_NUM_CTX
        other_opts = {k: v for k, v in extra_opts.items() if k != "num_ctx"}
        if other_opts:
            kwargs["extra_body"] = {"options": other_opts}

    # 3. Local vLLM / OpenAI-compatible
    elif provider in ("vllm", "openai_compatible"):
        default_url = "http://localhost:8000/v1"
        raw_url = (
            agent_cfg.get("base_url")
            or global_cfg.get("llm_base_url")
            or default_url
        )
        kwargs["api_base"] = _format_base_url(raw_url, provider)
        kwargs["api_key"] = (
            agent_cfg.get("api_key")
            or get_secure_api_key("llm_api_key", ("LLM_API_KEY",))
            or global_cfg.get("llm_api_key")
            or "dummy"
        )

    # Temperature (opt-in; None omitted)
    if "temperature" in agent_cfg and agent_cfg["temperature"] is not None:
        kwargs["temperature"] = float(agent_cfg["temperature"])
    elif (temp := global_cfg.get("llm_temperature")) is not None:
        kwargs["temperature"] = float(temp)

    # Max tokens
    if "max_tokens" in agent_cfg and agent_cfg["max_tokens"] is not None:
        kwargs["max_tokens"] = int(agent_cfg["max_tokens"])
    else:
        if (max_tok := global_cfg.get("llm_max_tokens")) is not None:
            kwargs["max_tokens"] = int(max_tok)
        else:
            kwargs["max_tokens"] = DEFAULT_LLM_MAX_TOKENS

        if provider in ("vllm", "openai_compatible") and kwargs.get("max_tokens") == 16384:
            kwargs["max_tokens"] = 4096

    if provider in ("ollama", "vllm", "openai_compatible"):
        kwargs["drop_params"] = True

    # Retries
    if "num_retries" in agent_cfg and agent_cfg["num_retries"] is not None:
        kwargs["num_retries"] = int(agent_cfg["num_retries"])
    elif (num_retries := global_cfg.get("llm_num_retries")) is not None:
        kwargs["num_retries"] = int(num_retries)
    else:
        kwargs["num_retries"] = DEFAULT_LLM_NUM_RETRIES

    return LiteLlm(model=model_name, **kwargs)

fetch_and_cache_openrouter_models async

fetch_and_cache_openrouter_models(target_type='llm', force_refresh=False, zotero_data_dir=None)

Fetch OpenRouter model list (LLM or Embedding), update memory & disk cache, and return full raw data.

Parameters:

Name	Type	Description	Default
target_type	str	'llm' or 'embedding'.	'llm'
force_refresh	bool	If True, bypass cache and perform live network request.	False
zotero_data_dir	Path | str | None	Optional Zotero data directory path.	None

Returns:

Type	Description
list[dict]	List of raw model dicts from OpenRouter API.

Source code in zocrates/llm.py

async def fetch_and_cache_openrouter_models(
        target_type: str = "llm",
        force_refresh: bool = False,
        zotero_data_dir: Path | str | None = None,
) -> list[dict]:
    """Fetch OpenRouter model list (LLM or Embedding), update memory & disk cache, and return full raw data.

    Args:
        target_type: 'llm' or 'embedding'.
        force_refresh: If True, bypass cache and perform live network request.
        zotero_data_dir: Optional Zotero data directory path.

    Returns:
        List of raw model dicts from OpenRouter API.
    """
    global _OPENROUTER_LLM_CACHE, _OPENROUTER_EMBED_CACHE, _OPENROUTER_DECISION_CACHE
    if target_type == "embedding":
        cache_dict = _OPENROUTER_EMBED_CACHE
    elif target_type == "decision":
        cache_dict = _OPENROUTER_DECISION_CACHE
    else:
        cache_dict = _OPENROUTER_LLM_CACHE

    now = time.time()

    # 1. Use memory cache if fresh (< 24 hours) and not forced
    if not force_refresh and cache_dict["data"] and (now - cache_dict["timestamp"] < CACHE_TTL_SECONDS):
        return cache_dict["data"]

    # 2. Try disk cache if not forcing refresh
    cache_file = None
    try:
        cache_file = _get_cache_file_path(target_type, zotero_data_dir)
        if not force_refresh and cache_file.exists():
            with open(cache_file, "r", encoding="utf-8") as f:
                cached = json.load(f)
                if now - cached.get("timestamp", 0) < CACHE_TTL_SECONDS:
                    cache_dict.clear()
                    cache_dict.update(cached)
                    return cached.get("data", [])
    except Exception as e:
        logger.debug("Disk cache unavailable (%s): %s", target_type, e)

    # 3. Network fetch requires API key
    key = get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))
    if not key:
        logger.info("No OPENROUTER_API_KEY configured; skipping network model fetch.")
        return cache_dict.get("data", [])

    # 3. Perform network fetch off the main event loop
    try:
        data = await asyncio.to_thread(_sync_fetch_openrouter_models, key, target_type)
        updated_cache = {"timestamp": now, "data": data}
        cache_dict.clear()
        cache_dict.update(updated_cache)

        # Write full raw endpoint response data to disk cache if available
        if cache_file is not None:
            try:
                with open(cache_file, "w", encoding="utf-8") as f:
                    json.dump(updated_cache, f, indent=2)
            except Exception as disk_err:
                logger.debug("Failed to write disk cache (%s): %s", target_type, disk_err)

        # Invalidate in-memory rate cache so fresh pricing takes effect immediately
        if target_type == "llm":
            try:
                from zocrates.utils.cost_calculator import clear_rate_cache
                clear_rate_cache()
            except Exception:
                pass

        return data
    except Exception as e:
        logger.error("Failed to fetch OpenRouter models (%s): %s", target_type, e)
        return cache_dict.get("data", [])

format_llm_exception

format_llm_exception(e, provider='openrouter_llm', base_url=None)

Format low-level litellm/httpx connection and API errors into actionable messages.

Parameters:

Name	Type	Description	Default
e	Exception	Caught exception.	required
provider	str	Provider identifier ('openrouter_llm', 'ollama', 'vllm', 'openai_compatible').	'openrouter_llm'
base_url	str | None	Base URL configured for local providers.	None

Returns:

Type	Description
str	User-friendly error message string.

Source code in zocrates/llm.py

def format_llm_exception(
    e: Exception,
    provider: str = "openrouter_llm",
    base_url: str | None = None,
) -> str:
    """Format low-level litellm/httpx connection and API errors into actionable messages.

    Args:
        e: Caught exception.
        provider: Provider identifier ('openrouter_llm', 'ollama', 'vllm', 'openai_compatible').
        base_url: Base URL configured for local providers.

    Returns:
        User-friendly error message string.
    """
    err_str = str(e)
    err_repr = repr(e)
    type_name = type(e).__name__

    is_conn_error = (
        isinstance(e, (ConnectionError, ConnectionRefusedError))
        or "ConnectError" in type_name
        or "APIConnectionError" in type_name
        or "ServiceUnavailableError" in type_name
        or any(
            term in err_str or term in err_repr
            for term in (
                "ConnectError",
                "ConnectionRefused",
                "APIConnectionError",
                "ServiceUnavailableError",
                "Connection refused",
                "Failed to establish a new connection",
                "Failed to connect",
            )
        )
    )

    if is_conn_error:
        if provider == "ollama":
            url = base_url or "http://localhost:11434"
            return (
                f"Unable to connect to Ollama server at {url}. "
                "Please verify Ollama is installed and running (`ollama serve`)."
            )
        elif provider in ("vllm", "openai_compatible"):
            url = base_url or "http://localhost:8000/v1"
            return (
                f"Unable to connect to local server at {url}. "
                "Please verify vLLM or custom OpenAI server is running with tool choice enabled (`--enable-auto-tool-choice --tool-call-parser hermes`)."
            )
        elif provider in ("openrouter", "openrouter_llm"):
            return (
                "Unable to connect to OpenRouter API. "
                "Please check your internet connection or verify your OPENROUTER_API_KEY."
            )

    if (
        "AuthenticationError" in type_name
        or "401" in err_str
        or "Unauthorized" in err_str
        or "invalid_api_key" in err_str.lower()
    ):
        return "Authentication failed (401). Please check your API key in Zocrates settings."

    return err_str

get_llm_model

get_llm_model(zotero_data_dir=None)

Instantiate the active ADK BaseLlm instance based on Zocrates configuration.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Optional path to Zotero data directory.	None

Returns:

Type	Description
BaseLlm	Configured ADK BaseLlm (LiteLlm) instance.

Source code in zocrates/llm.py

def get_llm_model(zotero_data_dir: Path | str | None = None) -> BaseLlm:
    """Instantiate the active ADK BaseLlm instance based on Zocrates configuration.

    Args:
        zotero_data_dir: Optional path to Zotero data directory.

    Returns:
        Configured ADK BaseLlm (LiteLlm) instance.
    """
    return build_agent_llm({}, zotero_data_dir=zotero_data_dir)

get_ollama_num_ctx

get_ollama_num_ctx(zotero_data_dir=None, cfg=None)

Return configured Ollama context window size (num_ctx), defaulting to 16384.

Source code in zocrates/config.py

def get_ollama_num_ctx(
    zotero_data_dir: Path | str | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return configured Ollama context window size (num_ctx), defaulting to 16384."""
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    val = cfg.get("ollama_num_ctx")
    if val is not None:
        try:
            return int(val)
        except (ValueError, TypeError):
            pass
    return DEFAULT_OLLAMA_NUM_CTX

get_openrouter_models async

get_openrouter_models(free_only=False, modality=None, min_context=0, target_type='llm', force_refresh=False, zotero_data_dir=None)

Filter session-cached OpenRouter models for LLMs or Cloud Embeddings.

Filters out unneeded modalities: audio, video, rerank, speech, and transcription.

Source code in zocrates/llm.py

async def get_openrouter_models(
        free_only: bool = False,
        modality: str | None = None,
        min_context: int = 0,
        target_type: str = "llm",
        force_refresh: bool = False,
        zotero_data_dir: Path | str | None = None,
) -> list[dict]:
    """Filter session-cached OpenRouter models for LLMs or Cloud Embeddings.

    Filters out unneeded modalities: audio, video, rerank, speech, and transcription.
    """
    raw_models = await fetch_and_cache_openrouter_models(
        target_type=target_type,
        force_refresh=force_refresh,
        zotero_data_dir=zotero_data_dir,
    )

    if target_type == "decision":
        decision_models = []
        for m in raw_models:
            model_id = m.get("id", "")
            if not model_id:
                continue
            arch = m.get("architecture") or {}
            out_mods = arch.get("output_modalities") or []
            if out_mods and "decisions" not in out_mods:
                continue
            model_name = m.get("name") or model_id
            context_len = m.get("context_length", 0)
            if min_context and context_len < min_context:
                continue
            description = m.get("description", "")
            pricing = m.get("pricing", {}) or {}
            try:
                p_prompt = float(pricing.get("prompt", -1))
                p_comp = float(pricing.get("completion", -1))
                is_free = model_id.endswith(":free") or (p_prompt == 0.0 and p_comp == 0.0)
            except (ValueError, TypeError):
                is_free = model_id.endswith(":free")
            if free_only and not is_free:
                continue
            decision_models.append(
                {
                    "id": model_id,
                    "name": model_name,
                    "context_length": context_len,
                    "description": description,
                    "pricing": pricing,
                    "architecture": arch,
                    "is_free": is_free,
                }
            )
        decision_models.sort(key=lambda x: x["name"].lower())
        return decision_models

    filtered = []

    EXCLUDED_MODALITIES = {"audio", "video", "speech", "transcription", "rerank"}

    for m in raw_models:
        model_id = m.get("id", "")
        model_name = m.get("name", model_id)
        context_len = m.get("context_length", 0)
        arch = m.get("architecture", {}) or {}
        input_mods = m.get("input_modalities") or arch.get("input_modalities") or []
        output_mods = m.get("output_modalities") or arch.get("output_modalities") or []
        pricing = m.get("pricing", {}) or {}

        if not input_mods or not output_mods:
            mod_str = arch.get("modality") or m.get("modality") or ""
            if "->" in mod_str:
                in_part, out_part = mod_str.split("->", 1)
                if not input_mods:
                    input_mods = [x.strip() for x in in_part.split("+") if x.strip()]
                if not output_mods:
                    output_mods = [x.strip() for x in out_part.split("+") if x.strip()]
            elif mod_str and not input_mods:
                input_mods = [x.strip() for x in mod_str.split("+") if x.strip()]

        if not input_mods:
            input_mods = ["text"]
        if not output_mods:
            output_mods = ["text"]

        # Modality checks
        if target_type == "embedding":
            if "text" not in input_mods:
                continue
        else:
            if "text" not in input_mods or "text" not in output_mods:
                continue

        id_lower = model_id.lower()
        # Strict Batch model exclusion
        if (
            "batch" in id_lower
            or "batch" in model_name.lower()
            or id_lower.endswith(":batch")
            or "(batch)" in model_name.lower()
        ):
            continue

        # Exclude non-LLM models (rerank, whisper, tts)
        if any(
            term in id_lower
            for term in (
                "rerank",
                "whisper",
                "tts",
            )
        ):
            continue

        def _is_zero_price(val: Any) -> bool:
            """Check whether a pricing value is zero.

            Args:
                val: Pricing scalar (string, float, int) to test.

            Returns:
                True if value evaluates to numeric 0.0, False otherwise.
            """
            try:
                return float(val) == 0.0
            except (ValueError, TypeError):
                return False

        is_free = model_id.endswith(":free") or (
                _is_zero_price(pricing.get("prompt")) and _is_zero_price(pricing.get("completion"))
        )
        if free_only and not is_free:
            continue
        if modality and modality not in input_mods:
            continue
        if min_context and context_len < min_context:
            continue

        formatted_id = f"openrouter/{model_id}" if not model_id.startswith("openrouter/") else model_id

        filtered.append(
            {
                "id": formatted_id,
                "name": model_name,
                "context_length": context_len,
                "input_modalities": input_mods,
                "output_modalities": output_mods,
                "architecture": arch,
                "pricing": pricing,
                "is_free": is_free,
            }
        )
    filtered.sort(key=lambda x: x["name"].lower())
    return filtered

get_secure_api_key

get_secure_api_key(key_name='openrouter_api_key', env_vars=('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY'))

Load a named API key from in-memory session environment variables.

Zotero Preferences / Login Manager serves as the secure persistent vault. Backend keys are in-memory session variables (os.environ) synced from Zotero.

Parameters:

Name	Type	Description	Default
key_name	str	Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').	'openrouter_api_key'
env_vars	list[str] | tuple[str, ...] | None	Environment variable names to check.	('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY')

Returns:

Type	Description
str | None	The API key string or None if not found.

Source code in zocrates/config.py

def get_secure_api_key(
    key_name: str = "openrouter_api_key",
    env_vars: list[str] | tuple[str, ...] | None = ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"),
) -> str | None:
    """Load a named API key from in-memory session environment variables.

    Zotero Preferences / Login Manager serves as the secure persistent vault.
    Backend keys are in-memory session variables (os.environ) synced from Zotero.

    Args:
        key_name: Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').
        env_vars: Environment variable names to check.

    Returns:
        The API key string or None if not found.
    """
    if env_vars:
        for var in env_vars:
            val = os.getenv(var)
            if val:
                return val
    return None

get_zocrates_data_dir

get_zocrates_data_dir(zotero_data_dir=None)

Return (and create if needed) the Zocrates data directory.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory (Path, string, or None for auto-discovery).	None

Returns:

Type	Description
Path	Path to <zotero_data_dir>/zocrates-data/, created if absent.

Source code in zocrates/config.py

def get_zocrates_data_dir(zotero_data_dir: Path | str | None = None) -> Path:
    """Return (and create if needed) the Zocrates data directory.

    Args:
        zotero_data_dir: Root Zotero data directory (Path, string, or None for auto-discovery).

    Returns:
        Path to ``<zotero_data_dir>/zocrates-data/``, created if absent.
    """
    resolved = resolve_zotero_data_dir(zotero_data_dir)
    if resolved is None:
        raise RuntimeError(
            "Cannot determine Zotero data directory. "
            "Please ensure Zotero is running and registered, or set ZOTERO_DATA_DIR."
        )
    zocrates_dir = resolved / ZOCRATES_DATA_SUBDIR
    zocrates_dir.mkdir(parents=True, exist_ok=True)
    return zocrates_dir

is_model_vision_capable

is_model_vision_capable(model_name, provider='openrouter_llm', zotero_data_dir=None)

Determine whether an LLM model identifier natively supports vision inputs.

Checks: 1. Tier 1: Authoritative OpenRouter catalog lookup (in-memory or disk cache). If found, modality metadata determines capability authoritatively. 2. Tier 2: Structural vision indicators (keywords 'vision', 'multimodal', 'vlm', and '-vl' suffixes). 3. Tier 3: Multimodal model families (Gemini, Claude 3+, GPT-4+, OpenAI o-series, GLM vision, Grok, Amazon Nova, Qwen flash, Pixtral, LLaVA, Omni, Molmo, InternVL, MiniCPM, CogVLM, Florence, Chameleon, Aria, PaliGemma). 4. Tier 4: LiteLLM capability fallback (only for non-OpenRouter providers).

Source code in zocrates/llm.py

def is_model_vision_capable(
    model_name: str,
    provider: str = "openrouter_llm",
    zotero_data_dir: Path | str | None = None,
) -> bool:
    """Determine whether an LLM model identifier natively supports vision inputs.

    Checks:
    1. Tier 1: Authoritative OpenRouter catalog lookup (in-memory or disk cache).
       If found, modality metadata determines capability authoritatively.
    2. Tier 2: Structural vision indicators (keywords 'vision', 'multimodal', 'vlm', and '-vl' suffixes).
    3. Tier 3: Multimodal model families (Gemini, Claude 3+, GPT-4+, OpenAI o-series, GLM vision,
       Grok, Amazon Nova, Qwen flash, Pixtral, LLaVA, Omni, Molmo, InternVL, MiniCPM, CogVLM,
       Florence, Chameleon, Aria, PaliGemma).
    4. Tier 4: LiteLLM capability fallback (only for non-OpenRouter providers).
    """
    if not model_name:
        return False

    clean_model = model_name.lower().strip()
    bare_model = clean_model[len("openrouter/"):] if clean_model.startswith("openrouter/") else clean_model
    is_openrouter = clean_model.startswith("openrouter/") or (
        bool(provider) and provider.lower() in ("openrouter", "openrouter_llm")
    )

    # Tier 1: Authoritative OpenRouter Catalog Lookup
    if is_openrouter:
        target_ids = {clean_model, bare_model, f"openrouter/{bare_model}"}
        catalog_entry = None

        # Check memory cache _OPENROUTER_LLM_CACHE
        raw_models = (
            _OPENROUTER_LLM_CACHE.get("data", [])
            if isinstance(_OPENROUTER_LLM_CACHE, dict)
            else (_OPENROUTER_LLM_CACHE if isinstance(_OPENROUTER_LLM_CACHE, list) else [])
        )
        if isinstance(raw_models, list):
            for m in raw_models:
                if isinstance(m, dict):
                    mid = str(m.get("id") or "").lower().strip()
                    mid_bare = mid[len("openrouter/"):] if mid.startswith("openrouter/") else mid
                    if mid in target_ids or mid_bare in target_ids:
                        catalog_entry = m
                        break

        # If empty or model not in memory cache, read disk cache
        if catalog_entry is None:
            try:
                candidates: list[Path] = []
                try:
                    candidates.append(_get_cache_file_path("llm", zotero_data_dir))
                except Exception:
                    pass
                if zotero_data_dir:
                    p = Path(zotero_data_dir)
                    candidates.append(p / "openrouter_models_cache" / "llm_models.json")
                    candidates.append(p / "zocrates-data" / "openrouter_models_cache" / "llm_models.json")

                disk_models = []
                for cache_path in candidates:
                    if cache_path.exists():
                        with open(cache_path, "r", encoding="utf-8") as f:
                            cached_data = json.load(f)
                        disk_models = (
                            cached_data.get("data", [])
                            if isinstance(cached_data, dict)
                            else (cached_data if isinstance(cached_data, list) else [])
                        )
                        if disk_models:
                            # Populate in-memory cache if empty
                            if (
                                isinstance(_OPENROUTER_LLM_CACHE, dict)
                                and not _OPENROUTER_LLM_CACHE.get("data")
                                and isinstance(cached_data, dict)
                            ):
                                _OPENROUTER_LLM_CACHE.clear()
                                _OPENROUTER_LLM_CACHE.update(cached_data)
                            break

                if isinstance(disk_models, list):
                    for m in disk_models:
                        if isinstance(m, dict):
                            mid = str(m.get("id") or "").lower().strip()
                            mid_bare = mid[len("openrouter/"):] if mid.startswith("openrouter/") else mid
                            if mid in target_ids or mid_bare in target_ids:
                                catalog_entry = m
                                break
            except Exception as disk_err:
                logger.debug("Failed reading OpenRouter disk cache in is_model_vision_capable: %s", disk_err)

        if catalog_entry is not None:
            arch = catalog_entry.get("architecture") if isinstance(catalog_entry.get("architecture"), dict) else {}
            input_modalities = catalog_entry.get("input_modalities") or arch.get("input_modalities")
            modality = catalog_entry.get("modality") or arch.get("modality")

            has_image = False
            if isinstance(input_modalities, (list, tuple, set)):
                has_image = any("image" in str(x).lower() for x in input_modalities)
            elif isinstance(input_modalities, str):
                has_image = "image" in input_modalities.lower()

            if not has_image and modality:
                if isinstance(modality, str):
                    has_image = "image" in modality.lower()
                elif isinstance(modality, (list, tuple, set)):
                    has_image = any("image" in str(x).lower() for x in modality)

            if has_image:
                return True
            if input_modalities is not None or modality is not None:
                # Explicitly in catalog with declared modalities but lacking 'image' -> authoritative False
                return False
            # If entry has no modality metadata at all, fall through to structural heuristics

    # Local Ollama declared capabilities probe
    is_ollama = clean_model.startswith(("ollama/", "ollama_chat/")) or (
        bool(provider) and provider.lower() == "ollama"
    )
    if is_ollama:
        try:
            cfg = load_zocrates_config(zotero_data_dir)
            b_url = (cfg.get("llm_base_url") or "http://localhost:11434").rstrip("/")
            bare_tag = bare_model.split("/", 1)[-1] if "/" in bare_model else bare_model
            bare_tag_base = bare_tag.split(":", 1)[0] if ":" in bare_tag else bare_tag

            # Check /api/tags
            try:
                resp = requests.get(f"{b_url}/api/tags", timeout=0.3)
                if resp.status_code == 200:
                    data = resp.json()
                    for m in data.get("models", []):
                        m_name = str(m.get("name") or m.get("model") or "").lower()
                        m_base = m_name.split(":", 1)[0] if ":" in m_name else m_name
                        if bare_tag in (m_name, m_base) or bare_tag_base in (m_name, m_base):
                            caps = [str(c).lower() for c in m.get("capabilities", [])]
                            details = m.get("details", {})
                            family = str(details.get("family", "")).lower()
                            families = [str(f).lower() for f in details.get("families", [])]
                            if "vision" in caps or "clip" in family or any("clip" in f for f in families):
                                return True
            except Exception:
                pass

            # Check /api/show
            try:
                resp = requests.post(f"{b_url}/api/show", json={"name": bare_tag}, timeout=0.3)
                if resp.status_code == 200:
                    data = resp.json()
                    caps = [str(c).lower() for c in (data.get("capabilities") or [])] if isinstance(data.get("capabilities"), list) else []
                    details = data.get("details", {})
                    family = str(details.get("family", "")).lower()
                    families = [str(f).lower() for f in details.get("families", [])]
                    if "vision" in caps or "clip" in family or any("clip" in f for f in families):
                        return True
            except Exception:
                pass
        except Exception as e:
            logger.debug("Failed querying Ollama for vision capability: %s", e)

    # Tier 2: Structural vision indicators (affixes, keywords, tags)
    if any(kw in clean_model for kw in ("vision", "multimodal", "vlm")):
        return True

    # Generalized structural regex: match any vl tag regardless of separator boundary
    # (matches qwen2.5vl, qwen3-vl, qwen4vl, deepseek-vl2, phi4vl, etc.)
    if re.search(r"(?:[-_./]|\b|[0-9a-z])vl\d*(?:[-_.:]|$|\b)", clean_model):
        return True

    # Tier 3: Multimodal model families (version-generalized)
    multimodal_family_patterns = (
        r"gemini",                                      # All Google Gemini models
        r"claude-.*(?:[3-9]|\d{2,})",                   # Anthropic Claude 3, 4, 5, 6 and beyond
        r"gpt-(?:[4-9]|\d{2,})",                        # OpenAI GPT-4, 4o, 4.5, 5, 6 and all subsequent generations
        r"(?:^|/)(?:o[1-9]|o\d{2,})(?!-mini)(?:[-_:]|$)", # OpenAI multimodal o-series (o1, o3, o4, o6...)
        r"(?:^|[/_-])glm-.*(?:flash|v\b)",              # GLM vision models (glm-5.3-flash, glm-4.5v, glm-5v-turbo, etc.)
        r"(?:^|[/_-])grok-.*(?:vision|[4-9]|\d{2,})",   # Grok vision models (grok-vision, grok-4.7, etc.)
        r"(?:^|[/_-])nova-(?:lite|pro|premier)",        # Amazon Nova models (nova-lite, nova-pro, nova-premier)
        r"(?:^|[/_-])gemma[-_]?(?:[3-9]|\d{2,})n?(?![-_:](?:1b|270m)(?:[-_:]|$))(?:[-_:]|$)", # Gemma 3 (>=4B), Gemma 4, and beyond
        r"paligemma",                                   # Google PaliGemma family
        r"qwen.*(?:vl|flash)",                          # Qwen VL and flash vision generations
        r"phi-.*(?:vision|multimodal|vl\b)",            # Microsoft Phi vision & multimodal
        r"mistral-.*(?:pixtral|vision|vl\b)",           # Mistral Pixtral & vision
        r"deepseek-.*(?:vl|vision)",                    # DeepSeek-VL family
        r"pixtral",                                     # Mistral Pixtral
        r"llava",                                       # LLaVA family
        r"omni",                                        # Omni multimodal models (e.g., Nemotron-Nano-Omni)
        r"molmo",                                       # AllenAI Molmo
        r"internvl",                                    # InternVL family
        r"minicpm",                                     # MiniCPM-V family
        r"cogvlm",                                      # CogVLM family
        r"florence",                                    # Microsoft Florence
        r"chameleon",                                   # Meta Chameleon
        r"aria",                                        # Rhymes Aria
    )

    if any(re.search(pat, clean_model) for pat in multimodal_family_patterns):
        return True

    # Tier 4: LiteLLM fallback (only for non-OpenRouter providers)
    if not is_openrouter:
        try:
            import litellm

            bare_name = clean_model.split("/")[-1] if "/" in clean_model else clean_model
            if litellm.supports_vision(model_name) or litellm.supports_vision(bare_name):
                return True
        except Exception:
            pass

    return False

load_zocrates_config

load_zocrates_config(zotero_data_dir=None)

Load the Zocrates config from the data directory.

If zotero_data_dir is None, attempts to obtain it from the active LibrarySyncer service or the ZOTERO_DATA_DIR environment variable.

Returns config merged with standard defaults.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory.	None

Returns:

Type	Description
dict	Parsed JSON config dict merged with defaults.

Source code in zocrates/config.py

def load_zocrates_config(zotero_data_dir: Path | str | None = None) -> dict:
    """Load the Zocrates config from the data directory.

    If ``zotero_data_dir`` is ``None``, attempts to obtain it from the active
    ``LibrarySyncer`` service or the ``ZOTERO_DATA_DIR`` environment variable.

    Returns config merged with standard defaults.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Parsed JSON config dict merged with defaults.
    """
    defaults = copy.deepcopy(DEFAULT_CONFIG)

    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return defaults

    config_path = get_config_path(resolved_dir)
    resolved_path = config_path.resolve()
    if not config_path.exists():
        with _CONFIG_CACHE_LOCK:
            _CONFIG_CACHE.pop(resolved_path, None)
            _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
        return defaults

    try:
        stat_res = config_path.stat()
        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
    except OSError:
        return defaults

    with _CONFIG_CACHE_LOCK:
        if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
            return copy.deepcopy(_CONFIG_CACHE[resolved_path])

    if config_path.exists():
        try:
            lock_path = str(config_path) + ".lock"
            with FileLock(lock_path):
                try:
                    stat_res = config_path.stat()
                    current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                except OSError:
                    return defaults

                with _CONFIG_CACHE_LOCK:
                    if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
                        return copy.deepcopy(_CONFIG_CACHE[resolved_path])

                with open(config_path) as f:
                    loaded = json.load(f)

                # Backfill override_agent_models if missing or incomplete
                needs_save = False

                if "override_agent_models" not in loaded or not isinstance(loaded["override_agent_models"], dict):
                    loaded["override_agent_models"] = copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS)
                    needs_save = True
                else:
                    for role, default_role_cfg in DEFAULT_OVERRIDE_AGENT_MODELS.items():
                        if role not in loaded["override_agent_models"]:
                            loaded["override_agent_models"][role] = copy.deepcopy(default_role_cfg)
                            needs_save = True

                # Backfill search_harness if missing or incomplete in existing config file on disk
                if "search_harness" not in loaded or not isinstance(loaded["search_harness"], dict):
                    loaded["search_harness"] = copy.deepcopy(DEFAULT_SEARCH_HARNESS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_SEARCH_HARNESS.items():
                        if k not in loaded["search_harness"]:
                            loaded["search_harness"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill graph_relations if missing or incomplete in existing config file on disk
                if "graph_relations" not in loaded or not isinstance(loaded["graph_relations"], dict):
                    loaded["graph_relations"] = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_GRAPH_RELATIONS.items():
                        if k not in loaded["graph_relations"]:
                            loaded["graph_relations"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill llm_num_retries if missing from existing config file on disk
                if "llm_num_retries" not in loaded:
                    loaded["llm_num_retries"] = DEFAULT_LLM_NUM_RETRIES
                    needs_save = True

                # Backfill llm_base_url if missing from existing config file on disk
                if "llm_base_url" not in loaded:
                    loaded["llm_base_url"] = DEFAULT_LLM_BASE_URL
                    needs_save = True

                # Backfill citations options if missing from existing config file on disk
                if "citations_enabled" not in loaded:
                    loaded["citations_enabled"] = DEFAULT_CITATIONS_ENABLED
                    needs_save = True
                if "references_section_enabled" not in loaded:
                    loaded["references_section_enabled"] = DEFAULT_REFERENCES_SECTION_ENABLED
                    needs_save = True
                if "considered_sources_disclosure" not in loaded:
                    loaded["considered_sources_disclosure"] = DEFAULT_CONSIDERED_SOURCES_DISCLOSURE
                    needs_save = True
                if "auto_render_mermaid_diagrams" not in loaded:
                    loaded["auto_render_mermaid_diagrams"] = DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS
                    needs_save = True
                if "enable_mcp_server" not in loaded:
                    loaded["enable_mcp_server"] = DEFAULT_ENABLE_MCP_SERVER
                    needs_save = True
                if "auto_spawn_backend" not in loaded:
                    loaded["auto_spawn_backend"] = DEFAULT_AUTO_SPAWN_BACKEND
                    needs_save = True
                if "enable_predictive_ghost_input" not in loaded:
                    loaded["enable_predictive_ghost_input"] = DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT
                    needs_save = True
                if "predictive_prompt_model" not in loaded:
                    loaded["predictive_prompt_model"] = DEFAULT_PREDICTIVE_PROMPT_MODEL
                    needs_save = True
                if "sentence_attribution_model" not in loaded:
                    loaded["sentence_attribution_model"] = DEFAULT_SENTENCE_ATTRIBUTION_MODEL
                    needs_save = True
                if "scholar_polite_email" not in loaded:
                    loaded["scholar_polite_email"] = DEFAULT_SCHOLAR_POLITE_EMAIL
                    needs_save = True
                if "conversion_timeout_seconds" not in loaded:
                    loaded["conversion_timeout_seconds"] = DEFAULT_CONVERSION_TIMEOUT_SECONDS
                    needs_save = True
                if "datalab_merge_cross_page" not in loaded:
                    loaded["datalab_merge_cross_page"] = DEFAULT_DATALAB_MERGE_CROSS_PAGE
                    needs_save = True
                if "user_research_profile" not in loaded:
                    loaded["user_research_profile"] = DEFAULT_USER_RESEARCH_PROFILE
                    needs_save = True
                if "max_research_profile_chars" not in loaded:
                    loaded["max_research_profile_chars"] = DEFAULT_MAX_RESEARCH_PROFILE_CHARS
                    needs_save = True
                if "max_saved_sessions" not in loaded:
                    loaded["max_saved_sessions"] = DEFAULT_MAX_SAVED_SESSIONS
                    needs_save = True
                if "session_retention_days" not in loaded:
                    loaded["session_retention_days"] = DEFAULT_SESSION_RETENTION_DAYS
                    needs_save = True
                if "session_db_auto_vacuum_on_startup" not in loaded:
                    loaded["session_db_auto_vacuum_on_startup"] = DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP
                    needs_save = True
                if "prune_empty_sessions" not in loaded:
                    loaded["prune_empty_sessions"] = DEFAULT_PRUNE_EMPTY_SESSIONS
                    needs_save = True
                if "backend_port" not in loaded:
                    loaded["backend_port"] = DEFAULT_BACKEND_PORT
                    needs_save = True

                # Backfill Phase 7 onboarding SSOT keys if missing from existing
                # config files on disk (embedding_dimension / sync_auto_abort_failures)
                if "embedding_dimension" not in loaded:
                    loaded["embedding_dimension"] = DEFAULT_CONFIG["embedding_dimension"]
                    needs_save = True
                if "sync_auto_abort_failures" not in loaded:
                    loaded["sync_auto_abort_failures"] = DEFAULT_SYNC_AUTO_ABORT_FAILURES
                    needs_save = True

                if "vlm_captioning_backup_model" not in loaded:
                    loaded["vlm_captioning_backup_model"] = DEFAULT_CONFIG["vlm_captioning_backup_model"]
                    needs_save = True
                if "vlm_captioning_model" not in loaded:
                    loaded["vlm_captioning_model"] = DEFAULT_CONFIG["vlm_captioning_model"]
                    needs_save = True
                if "vlm_captioning_provider" not in loaded:
                    loaded["vlm_captioning_provider"] = DEFAULT_CONFIG["vlm_captioning_provider"]
                    needs_save = True
                if "enable_vlm_captioning" not in loaded:
                    loaded["enable_vlm_captioning"] = DEFAULT_CONFIG["enable_vlm_captioning"]
                    needs_save = True
                if "vlm_max_images_per_doc" not in loaded:
                    loaded["vlm_max_images_per_doc"] = DEFAULT_CONFIG["vlm_max_images_per_doc"]
                    needs_save = True

                # Detect and migrate deprecated compaction keys from disk
                if "compaction_token_threshold" in loaded and "adaptive_compaction_token_ceiling" not in loaded:
                    loaded["adaptive_compaction_token_ceiling"] = loaded["compaction_token_threshold"]
                    needs_save = True
                if "enable_context_compaction" in loaded and "enable_adaptive_compaction" not in loaded:
                    loaded["enable_adaptive_compaction"] = loaded["enable_context_compaction"]
                    needs_save = True
                if "adaptive_decision_model" in loaded and "adaptive_compaction_decision_model" not in loaded:
                    loaded["adaptive_compaction_decision_model"] = loaded.pop("adaptive_decision_model")
                    needs_save = True
                elif "adaptive_decision_model" in loaded:
                    loaded.pop("adaptive_decision_model", None)
                    needs_save = True

                # Purge all 5 deprecated compaction keys
                deprecated_compaction_keys = (
                    "compaction_interval",
                    "compaction_overlap_size",
                    "compaction_token_threshold",
                    "compaction_event_retention_size",
                    "enable_context_compaction",
                )
                for dep_key in deprecated_compaction_keys:
                    if dep_key in loaded:
                        loaded.pop(dep_key, None)
                        needs_save = True

                # Backfill missing adaptive compaction options
                adaptive_compaction_defaults = {
                    "enable_adaptive_compaction": DEFAULT_ENABLE_ADAPTIVE_COMPACTION,
                    "adaptive_compaction_retention_turns": DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
                    "adaptive_compaction_min_turns": DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS,
                    "adaptive_compaction_token_floor": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR,
                    "adaptive_compaction_token_ceiling": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING,
                    "adaptive_compaction_threshold": DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD,
                    "adaptive_compaction_decision_model": DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL,
                    "compaction_summarizer_model": DEFAULT_COMPACTION_SUMMARIZER_MODEL,
                }
                for k, default_val in adaptive_compaction_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True

                # Disallow local Laya for compaction decisions on disk; reset to default remote model
                if is_local_decision_model(loaded.get("adaptive_compaction_decision_model")):
                    loaded["adaptive_compaction_decision_model"] = DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL
                    needs_save = True

                # Backfill citation verification options if missing from existing config file on disk
                verification_defaults = {
                    "citation_verification_enabled": DEFAULT_CITATION_VERIFICATION_ENABLED,
                    "citation_verification_provider": DEFAULT_CITATION_VERIFICATION_PROVIDER,
                    "citation_verification_model": DEFAULT_CITATION_VERIFICATION_MODEL,
                    "citation_verification_threshold": DEFAULT_CITATION_VERIFICATION_THRESHOLD,
                    "citation_verification_local_url": DEFAULT_CITATION_VERIFICATION_LOCAL_URL,
                    "citation_verification_local_port": DEFAULT_CITATION_VERIFICATION_LOCAL_PORT,
                    "citation_verification_auto_start_local": DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL,
                    "citation_verification_timeout_sec": DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC,
                    "citation_verification_turn_budget_sec": DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC,
                }
                for k, default_val in verification_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True
                # Migrate legacy default 3.0s/10.0s timeout and budget to 10.0s/20.0s for CPU inference headroom
                try:
                    if float(loaded.get("citation_verification_timeout_sec", 0)) == 3.0:
                        loaded["citation_verification_timeout_sec"] = DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass
                try:
                    if float(loaded.get("citation_verification_turn_budget_sec", 0)) == 10.0:
                        loaded["citation_verification_turn_budget_sec"] = DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass

                if needs_save:
                    try:
                        with open(config_path, "w") as f_out:
                            json.dump(loaded, f_out, indent=2)
                        stamp_config_mtime(resolved_dir)
                    except Exception as save_err:
                        logger.warning(
                            "Failed to backfill default dialogue agent models to %s: %s",
                            config_path,
                            save_err,
                        )
                    try:
                        stat_res = config_path.stat()
                        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                    except OSError:
                        pass

                res = dict(defaults)
                res.update(loaded)
                with _CONFIG_CACHE_LOCK:
                    _CONFIG_CACHE[resolved_path] = copy.deepcopy(res)
                    _CONFIG_CACHE_FINGERPRINT[resolved_path] = current_fingerprint
                return copy.deepcopy(res)
        except Exception as e:
            logger.warning("Failed to load config from %s: %s", config_path, e)
    return defaults

test_embedding_connection async

test_embedding_connection(provider=None, model=None, openrouter_api_key=None, zotero_data_dir=None)

Perform a lightweight test embedding request to verify embedding connectivity.

Source code in zocrates/llm.py

async def test_embedding_connection(
    provider: str | None = None,
    model: str | None = None,
    openrouter_api_key: str | None = None,
    zotero_data_dir: Path | str | None = None,
) -> dict:
    """Perform a lightweight test embedding request to verify embedding connectivity."""
    from zocrates.rag.embeddings import OpenRouterEmbedder

    cfg = load_zocrates_config(zotero_data_dir)
    provider = provider or cfg.get("embedding_provider", "local")
    model_name = model or cfg.get("embedding_model", "jinaai/jina-embeddings-v5-text-nano")

    if provider == "local":
        return {
            "ok": True,
            "status": "ok",
            "message": "Local Jina v5 Nano embedder ready",
            "model": model_name,
            "dimension": 768,
        }

    key = openrouter_api_key or get_secure_api_key(
        "openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY")
    )
    if not key:
        return {
            "ok": False,
            "status": "error",
            "message": "OPENROUTER_API_KEY is not configured.",
            "model": model_name,
        }

    try:
        def _run_embedding() -> int:
            """Embed a test string and return embedding dimension.

            Returns:
                Integer dimension of the test embedding vector.
            """
            embedder = OpenRouterEmbedder(model_name=model_name, api_key=key)
            vecs = embedder(["test"])
            dim = len(vecs[0]) if vecs and len(vecs) > 0 else 0
            return dim

        dim = await asyncio.to_thread(_run_embedding)
        return {
            "ok": True,
            "status": "ok",
            "message": f"Connected! (dimension: {dim})",
            "model": model_name,
            "dimension": dim,
        }
    except Exception as e:
        logger.error("Embedding connection test failed for %s: %s", model_name, e)
        user_msg = format_llm_exception(e, provider="openrouter_llm")
        return {
            "ok": False,
            "status": "error",
            "message": user_msg,
            "model": model_name,
        }

test_llm_connection async

test_llm_connection(provider=None, model=None, base_url=None, openrouter_api_key=None, zotero_data_dir=None)

Perform a lightweight test completion request to verify LLM connectivity.

Supports testing unsaved form parameters or defaults to saved config.

Source code in zocrates/llm.py

async def test_llm_connection(
    provider: str | None = None,
    model: str | None = None,
    base_url: str | None = None,
    openrouter_api_key: str | None = None,
    zotero_data_dir: Path | str | None = None,
) -> dict:
    """Perform a lightweight test completion request to verify LLM connectivity.

    Supports testing unsaved form parameters or defaults to saved config.
    """
    cfg = load_zocrates_config(zotero_data_dir)
    provider = provider or cfg.get("llm_provider", "openrouter_llm")
    raw_model = model or cfg.get("llm_model") or PROVIDER_DEFAULT_MODELS.get(provider, DEFAULT_LLM_MODEL)
    model_name = _resolve_model_name(raw_model, provider)

    kwargs: Any = {"max_tokens": 5}

    if provider in ("openrouter", "openrouter_llm"):
        key = openrouter_api_key or get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))
        if not key:
            return {
                "ok": False,
                "status": "error",
                "message": "OPENROUTER_API_KEY is not configured.",
                "model": model_name,
            }
        kwargs["api_key"] = key
        kwargs["extra_headers"] = {
            # "HTTP-Referer": "https://github.com/alexplaka/Zocrates",
            "X-OpenRouter-Title": "Zocrates",
        }

    elif provider == "ollama":
        raw_url = base_url or cfg.get("llm_base_url") or "http://localhost:11434"
        kwargs["api_base"] = _format_base_url(raw_url, "ollama")
        kwargs["num_ctx"] = get_ollama_num_ctx(cfg=cfg)
        kwargs["drop_params"] = True

    elif provider in ("vllm", "openai_compatible"):
        raw_url = base_url or cfg.get("llm_base_url") or "http://localhost:8000/v1"
        kwargs["api_base"] = _format_base_url(raw_url, provider)
        kwargs["api_key"] = (
            get_secure_api_key("llm_api_key", ("LLM_API_KEY",))
            or cfg.get("llm_api_key")
            or "dummy"
        )
        kwargs["drop_params"] = True

    try:
        messages = [{"role": "user", "content": "Ping"}]
        response = await asyncio.to_thread(
            litellm.completion,
            model=model_name,
            messages=messages,
            **kwargs,
        )
        content = "OK"
        if response and getattr(response, "choices", None):
            choices = response.choices
            if isinstance(choices, (list, tuple)) and len(choices) > 0:
                first = choices[0]
                if isinstance(first, dict):
                    msg = first.get("message", {})
                    content = msg.get("content", "OK") if isinstance(msg, dict) else getattr(msg, "content", "OK")
                elif hasattr(first, "message"):
                    msg = getattr(first, "message", None)
                    if msg is not None:
                        content = getattr(msg, "content", "OK") if hasattr(msg, "content") else (msg.get("content", "OK") if isinstance(msg, dict) else "OK")
        if not content:
            content = "OK"

        # Tool capability probe for local endpoints (Ollama, vLLM, OpenAI-compatible)
        if provider in ("ollama", "vllm", "openai_compatible"):
            try:
                dummy_tools = [
                    {
                        "type": "function",
                        "function": {
                            "name": "ping_tool",
                            "description": "Diagnostic probe verifying function calling / tool support.",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "query": {
                                        "type": "string",
                                        "description": "Probe query parameter.",
                                    }
                                },
                                "required": ["query"],
                            },
                        },
                    }
                ]
                probe_kwargs = dict(kwargs)
                probe_kwargs["tools"] = dummy_tools
                probe_kwargs["tool_choice"] = "auto"
                await asyncio.to_thread(
                    litellm.completion,
                    model=model_name,
                    messages=[{"role": "user", "content": "Ping"}],
                    **probe_kwargs,
                )
            except Exception as tool_exc:
                logger.warning("Tool capability probe failed for %s (%s): %s", model_name, provider, tool_exc)
                if provider in ("vllm", "openai_compatible"):
                    return {
                        "ok": False,
                        "status": "error",
                        "message": (
                            f"Tool calling probe failed: {tool_exc}. "
                            "Please verify your local server is running with tool calling enabled "
                            "(for vLLM: --enable-auto-tool-choice --tool-call-parser hermes)."
                        ),
                        "model": model_name,
                    }
                else:
                    return {
                        "ok": False,
                        "status": "error",
                        "message": f"Ollama tool calling probe failed: {tool_exc}",
                        "model": model_name,
                    }

        return {
            "ok": True,
            "status": "ok",
            "message": f"Connection successful! Response: {str(content)[:30]}",
            "model": model_name,
        }
    except Exception as e:
        logger.error("LLM connection test failed for %s: %s", model_name, e)
        base_url = kwargs.get("api_base")
        user_msg = format_llm_exception(e, provider=provider, base_url=base_url)
        return {
            "ok": False,
            "status": "error",
            "message": user_msg,
            "model": model_name,
        }