Function _llm_call

openkb/agent/compiler.py:400–431

Single LLM call with animated progress and debug logging.

(
    model: str, messages: list[dict], step_name: str, raise_on_truncation: bool = False, **kwargs
)

def _llm_call(
    model: str, messages: list[dict], step_name: str, raise_on_truncation: bool = False, **kwargs
) -> str:
    """Single LLM call with animated progress and debug logging."""
    messages = _prepare_messages(model, messages)
    extra_headers = get_extra_headers()
    if extra_headers:
        kwargs.setdefault("extra_headers", extra_headers)
    timeout = get_timeout()
    if timeout is not None:
        kwargs.setdefault("timeout", timeout)
    logger.debug("LLM request [%s]:\n%s", step_name, _fmt_messages(messages))
    if kwargs:
        logger.debug("LLM kwargs [%s]: %s", step_name, kwargs)

    spinner = _Spinner(step_name)
    spinner.start()
    t0 = time.time()

    response = litellm.completion(model=model, messages=messages, **kwargs)
    content = response.choices[0].message.content or ""
    truncated = _warn_if_truncated(response, step_name, kwargs.get("max_tokens"))

    spinner.stop(_format_usage(time.time() - t0, response.usage))
    logger.debug(
        "LLM response [%s]:\n%s", step_name, content[:500] + ("..." if len(content) > 500 else "")
    )
    if raise_on_truncation and truncated:
        raise TruncatedResponseError(
            f"LLM [{step_name}] hit the length limit; skipping to avoid a truncated page"
        )
    return content.strip()


async def _llm_call_async(