changedetection.io/changedetectionio/llm/client.py

"""
Thin wrapper around litellm.completion.
Keeps litellm import isolated so the rest of the codebase doesn't depend on it directly,
and makes the call easy to mock in tests.
"""

import logging
import os

from loguru import logger

# Default output token cap for JSON-returning calls (intent eval, preview, setup).
# These return small JSON objects — 400 is enough for a verbose explanation while
# still preventing runaway cost. Change summaries pass their own max_tokens via
# _summary_max_tokens() and are NOT subject to this cap.
_MAX_COMPLETION_TOKENS = 400

# Default request timeout (seconds). Raised from 60 to 300 because even cloud
# reasoning models can be slow on the first hit (issue #4225). Overridable via
# LLM_TIMEOUT.
DEFAULT_TIMEOUT = int(os.getenv('LLM_TIMEOUT', 300))
# Relaxed timeout for local / self-hosted endpoints (Ollama, vLLM, LM Studio,
# llama.cpp on localhost or a LAN address). These run on modest hardware and can
# spend many minutes on prompt prefill before the first token, so they get a much
# longer deadline (Hermes-style, 30 min). Overridable via LLM_LOCAL_TIMEOUT; see
# evaluator.resolve_llm_timeout() for how the endpoint is classified.
DEFAULT_LOCAL_TIMEOUT = int(os.getenv('LLM_LOCAL_TIMEOUT', 1800))
# Models and reasoning architectures that reject explicit sampling parameters (temperature/top_p).
# Substring match against the lowercased model name, so 'gpt-5' also covers gpt-5-mini, gpt-5.1 etc.
#
# This is only an optimisation - the BadRequestError handler below strips and retries for
# anything not listed. Being listed just avoids paying a rejected round trip on every call.
#
# NOT all GPT models: gpt-4o, gpt-4-turbo and gpt-3.5-turbo take temperature normally. It is
# reasoning that forbids it. gpt-5 reports "only temperature=1 is supported unless
# reasoning_effort resolves to 'none'" - we never set reasoning_effort=none, so it is always
# in that state for us.
_NO_TEMPERATURE_MODEL_KEYWORDS = ('flash-lite', 'thinking-exp', 'o1', 'o3', 'o4', 'gpt-5')

DEFAULT_RETRIES = 3


class _LoguruInterceptHandler(logging.Handler):
    # Routes litellm's stdlib log records through loguru so debug output
    # uses the same format/sink as the rest of the app.
    def emit(self, record):
        try:
            level = logger.level(record.levelname).name
        except (ValueError, AttributeError):
            level = record.levelno
        logger.opt(exception=record.exc_info).log(level, record.getMessage())


_debug_installed = False


def _install_litellm_debug():
    # Attach our loguru intercept and clear any pre-existing handlers so litellm's
    # own stdout StreamHandler (installed by _turn_on_debug / set_verbose) doesn't
    # double-emit. Setting the logger level to DEBUG is enough to make litellm
    # produce debug records — we don't call _turn_on_debug() for that reason.
    global _debug_installed
    if _debug_installed:
        return

    handler = _LoguruInterceptHandler()
    handler.setLevel(logging.DEBUG)
    for _name in ('LiteLLM', 'litellm', 'litellm.utils', 'litellm.router'):
        _lg = logging.getLogger(_name)
        _lg.handlers = []
        _lg.setLevel(logging.DEBUG)
        _lg.addHandler(handler)
        _lg.propagate = False

    _debug_installed = True
    logger.info("LLM client: litellm debug logging routed through loguru")


def completion(  # noqa: C901
    model: str,
    messages: list,
    api_key: str = None,
    api_base: str = None,
    timeout: int = DEFAULT_TIMEOUT,
    max_tokens: int = None,
    extra_body: dict = None,
    debug: bool = False,
) -> tuple[str, int, int, int]:
    """
    Call the LLM and return (response_text, total_tokens, input_tokens, output_tokens).
    Retries up to DEFAULT_RETRIES times on timeout or connection errors.
    Token counts are 0 if the provider doesn't return usage data.
    Raises on network/auth errors — callers handle gracefully.

    timeout: seconds for the request. Local endpoints get a longer value than cloud —
    see evaluator.resolve_llm_timeout().
    """
    try:
        import litellm
    except ImportError:
        raise RuntimeError("litellm is not installed. Add it to requirements.txt.") from None

    if debug:
        _install_litellm_debug()

    _timeout = timeout if timeout is not None else DEFAULT_TIMEOUT

    kwargs = {
        'model': model,
        'messages': messages,
        'timeout': _timeout,
        'max_tokens': max_tokens if max_tokens is not None else _MAX_COMPLETION_TOKENS,
    }
    _m_lower = (model or '').lower()
    if not any(k in _m_lower for k in _NO_TEMPERATURE_MODEL_KEYWORDS):
        kwargs['temperature'] = 0

    if api_key:
        kwargs['api_key'] = api_key
    if api_base:
        kwargs['api_base'] = api_base
    if extra_body:
        kwargs['extra_body'] = extra_body

    _retryable = (litellm.Timeout, litellm.APIConnectionError)

    # Some models reject sampling params outright: Anthropic Claude Opus 4.7/4.8 and
    # Fable return HTTP 400 for 'temperature', and OpenAI reasoning models (o1/o3/gpt-5)
    # only accept the default. litellm's per-model param metadata lags new releases, so
    # drop_params can't be relied on for freshly released models — instead, if the provider
    # rejects a sampling param, strip them and retry once. Models that accept them are
    # unaffected (they still receive temperature=0).
    _sampling_params = ('temperature', 'top_p', 'top_k')
    _stripped_sampling = False

    logger.debug(
        f"LLM client: calling model={model!r} api_base={api_base!r} "
        f"timeout={_timeout}s max_tokens={kwargs['max_tokens']}"
    )
    logger.trace(messages)

    attempt = 0
    while attempt < DEFAULT_RETRIES:
        attempt += 1
        try:
            response = litellm.completion(**kwargs)
            choice = response.choices[0]
            message = choice.message
            finish = getattr(choice, 'finish_reason', None)

            text = message.content or ''

            if not text:
                # Some providers (e.g. Gemini) put text in message.parts instead of .content
                parts = getattr(message, 'parts', None)
                if parts:
                    text = ''.join(getattr(p, 'text', '') or '' for p in parts).strip()
                    logger.debug(
                        f"LLM client: extracted text from message.parts ({len(parts)} parts) model={model!r}"
                    )

            if finish == 'length':
                logger.warning(
                    f"LLM client: response truncated (finish_reason='length') model={model!r} "
                    f"— increase max_tokens; got {len(text)} chars so far"
                )

            if not text:
                logger.warning(
                    f"LLM client: empty content from model={model!r} "
                    f"finish_reason={finish!r} "
                    f"message={message!r}"
                )

            usage = getattr(response, 'usage', None)
            input_tokens = int(getattr(usage, 'prompt_tokens', 0) or 0) if usage else 0
            output_tokens = int(getattr(usage, 'completion_tokens', 0) or 0) if usage else 0
            total_tokens = (
                int(getattr(usage, 'total_tokens', 0) or 0)
                if usage
                else (input_tokens + output_tokens)
            )
            logger.debug(
                f"LLM client: model={model!r} finish={finish!r} "
                f"tokens={total_tokens} (in={input_tokens} out={output_tokens}) "
                f"text_len={len(text)}"
            )
            return text, total_tokens, input_tokens, output_tokens

        except _retryable as e:
            # litellm formats its Timeout message with None when the provider doesn't
            # propagate the timeout value — patch the exception args in-place so every
            # caller that logs str(e) sees the real number.
            _fix = f'after {_timeout} seconds'
            try:
                e.args = tuple(str(a).replace('after None seconds', _fix) for a in e.args)
            except Exception:
                pass
            if attempt < DEFAULT_RETRIES:
                logger.warning(
                    f"LLM call timed out/connection error (attempt {attempt}/{DEFAULT_RETRIES}), "
                    f"retrying — model={model!r} timeout={_timeout}s error={e}"
                )
                continue
            logger.warning(
                f"LLM call failed after {DEFAULT_RETRIES} attempts ({_timeout}s timeout) "
                f"model={model!r} error={e}"
            )
            raise

        except litellm.BadRequestError as e:
            # If the provider rejected an unsupported sampling param or extra_body
            # (e.g. Gemini INVALID_ARGUMENT on thinkingConfig or temperature), drop
            # them and retry once.
            if not _stripped_sampling:
                dropped = [p for p in _sampling_params if kwargs.pop(p, None) is not None]
                extra_body = kwargs.get('extra_body')
                if isinstance(extra_body, dict):
                    gen_cfg = extra_body.get('generationConfig')
                    if (
                        isinstance(gen_cfg, dict)
                        and gen_cfg.pop('thinkingConfig', None) is not None
                    ):
                        dropped.append('thinkingConfig')
                        if not gen_cfg:
                            extra_body.pop('generationConfig', None)
                        if not extra_body:
                            kwargs.pop('extra_body', None)
                    elif 'thinkingConfig' in extra_body:
                        extra_body.pop('thinkingConfig', None)
                        dropped.append('thinkingConfig')
                        if not extra_body:
                            kwargs.pop('extra_body', None)

                if dropped:
                    _stripped_sampling = True
                    attempt -= 1
                    logger.warning(
                        f"LLM client: model={model!r} rejected request ({e}); "
                        f"stripped {dropped} and retrying once"
                    )
                    continue
            logger.warning(f"LLM call failed: model={model!r} error={e}")
            raise

        except Exception as e:
            logger.warning(f"LLM call failed: model={model!r} error={e}")
            raise


def _completion_worker(conn, log_level):
    # Runs in a spawned child: litellm is imported here, and all of it (~200MB RSS) goes
    # back to the OS when the process exits.
    import json
    import sys
    logger.remove()
    logger.add(sys.stderr, level=log_level)
    try:
        kwargs = json.loads(conn.recv_bytes().decode('utf-8'))
        result = {'ok': True, 'result': list(completion(**kwargs))}
    except Exception as e:
        result = {'ok': False, 'error': str(e), 'error_type': type(e).__name__}
    try:
        conn.send_bytes(json.dumps(result).encode('utf-8'))
    finally:
        conn.close()


def completion_isolated(**kwargs):
    """Same as completion(), but runs it in a short-lived subprocess when litellm isn't loaded yet.

    For one-off calls like the settings "Test connection" button, so trying out a provider
    doesn't leave litellm resident for the life of the process. If litellm is already
    imported (AI is configured and in use) that memory is spent anyway, so just call
    completion() directly. Linux only, same as the extruct isolation in restock_diff.

    Errors from the child are re-raised as RuntimeError carrying the original message.
    """
    import json
    import platform
    import sys

    if 'litellm' in sys.modules or platform.system() != 'Linux':
        return completion(**kwargs)

    import multiprocessing
    try:
        log_level = logger._core.min_level
    except AttributeError:
        log_level = 'DEBUG'

    ctx = multiprocessing.get_context('spawn')
    parent_conn, child_conn = ctx.Pipe()
    p = ctx.Process(target=_completion_worker, args=(child_conn, log_level), daemon=True)
    p.start()
    child_conn.close()
    try:
        parent_conn.send_bytes(json.dumps(kwargs).encode('utf-8'))
        # completion() retries timeouts/connection errors, so allow for every attempt
        # plus the child's start-up (importing litellm takes a few seconds)
        _timeout = kwargs.get('timeout') or DEFAULT_TIMEOUT
        deadline = _timeout * (DEFAULT_RETRIES + 1) + 60
        if not parent_conn.poll(deadline):
            raise TimeoutError(f"LLM call did not finish within {deadline}s")
        try:
            result = json.loads(parent_conn.recv_bytes().decode('utf-8'))
        except EOFError:
            raise RuntimeError("LLM subprocess exited without a result, check the logs") from None
    finally:
        parent_conn.close()
        p.join(timeout=5)
        if p.is_alive():
            p.kill()
            p.join()

    if not result['ok']:
        raise RuntimeError(result['error'])
    return tuple(result['result'])