llm/warmup.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Prime the model server's prefix cache so the first question is not the slow one.
Every SQL-generation request starts with the same bytes: the static prompt
prefix (:func:`prompt_engine.static_prefix.build_static_prefix`) is
byte-identical per data source, so a model server that caches prompt
prefixes (vLLM with ``--enable-prefix-caching``, llama.cpp's prompt cache)
only pays prefill for the short per-question tail. The cache starts empty,
though. After a restart of this server, or of the model server, the first
question pays the full prefill of the whole prefix -- about a minute on a
large local model -- and nothing before this module made that cost land on
anything but a user's question.
:func:`warm_prefix_cache` sends the model server one minimal request per
data source: the messages are exactly the cacheable prefix the real requests
start with, ``max_tokens`` is ``1`` and the temperature ``0``
(:meth:`llm.providers.OpenAIBackend.warm_prefix`). The server computes the
prefix's KV cache to answer it and keeps it.
:func:`start_background_warmup` runs that on a daemon thread from
``api/server.py``'s ``lifespan``, so the server accepts requests at once; a
question that arrives while the warm-up is still running is served as it
would have been without one (and on a server that shares in-flight prefix
blocks, may even benefit from it).
Why the prefix is not re-implemented here
-----------------------------------------
The warm-up prompt is taken from :func:`llm.router.build_prompt_segments` --
the one function both engines build real prompts with -- called with an
empty question. Its ``static_prefix`` is therefore the real requests'
prefix by construction: it cannot drift from it when the template, the
per-source scoping or the join-only rules change. ``tests/test_llm_warmup.py``
asserts that a real request's flattened prompt starts with exactly the text
the warm-up sent.
Keeping it warm
---------------
The start-up pass primes the cache once. On a model server shared with other
users the cache is evicted again whenever their traffic needs the memory, and
the next question pays the whole prefill. :class:`KeepWarm` (started from
``api/server.py``'s ``lifespan`` when ``LLM_PREFIX_KEEPWARM_SECONDS`` is above
zero) repeats the same pass every that many seconds. It is not a second
implementation of the warm-up: every pass is :func:`warm_prefix_cache`, with
three differences that only apply to it --
* a source whose prefix a real request used within the last interval is
skipped (:mod:`llm.prefix_activity`; the request refreshed the cache, so
under traffic the keep-warm sends nothing);
* a repeated outcome is logged once, then at ``DEBUG`` until it changes (an
endpoint that stays down is one warning, not one per pass);
* it does nothing while ``paused()`` says so (maintenance mode) and stops at
shutdown.
Failure policy
--------------
A warm-up is an optimisation. Every failure -- an unreachable endpoint, a
timeout, a server that rejects ``max_tokens=1``, a bad ``LLM_EXTRA_BODY``,
an untrusted endpoint, a configuration error -- is logged as one line and
swallowed; nothing here can stop the server starting or a question being
answered. The lines never contain prompt text, response text or keys: the
exception's class and, for an HTTP error, its status code are all that is
said about a failure.
"""
from __future__ import annotations
import logging
import threading
import time
from collections.abc import Callable
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Any
import config as cfg
from llm.prefix_activity import PrefixActivity, activity as _default_activity
from llm.router import (
LLMRouter,
PromptSegments,
RemoteProviderNotAllowedError,
build_prompt_segments,
)
logger = logging.getLogger(__name__)
#: Held for the whole of one warm-up pass, so a startup warm-up and a
#: manual ``POST /admin/llm/warmup`` never send the same prefix twice at
#: once.
_warmup_lock = threading.Lock()
@dataclass(frozen=True)
class SourceWarmup:
"""What happened when one data source's prefix was warmed.
Parameters
----------
source:
The data source name, or ``None`` for the whole schema (a
deployment with one data source).
status:
``"warmed"``, ``"skipped"`` (nothing to warm, or not allowed to)
or ``"failed"``.
detail:
Why it was skipped or failed, in words that carry no prompt text;
``None`` for ``"warmed"``.
duration_ms:
Wall-clock time spent on the request, or ``None`` when none was
sent.
prompt_tokens:
``usage.prompt_tokens`` the server reported for the warm-up
request -- the size of the prefix in the model's own tokens.
cached_tokens:
How many of those the server already had cached, when it reports
that (vLLM's ``--enable-prompt-tokens-details``); a value close to
``prompt_tokens`` means the cache was already warm.
Examples
--------
>>> SourceWarmup("sales", "warmed", None, 41000, 4612, None).as_dict()["prompt_tokens"]
4612
"""
source: str | None
status: str
detail: str | None = None
duration_ms: int | None = None
prompt_tokens: int | None = None
cached_tokens: int | None = None
def as_dict(self) -> dict[str, Any]:
"""A JSON-ready copy, for ``POST /admin/llm/warmup``."""
return asdict(self)
def _source_label(source: str | None) -> str:
"""The name a log line uses for *source*."""
return source if source is not None else "(whole schema)"
def _describe_failure(exc: BaseException) -> str:
"""A log-safe description of *exc*: its class, plus the HTTP status if it has one.
Exception *messages* are left out on purpose. Those of ``requests``
carry URLs, and those of an endpoint's error body can echo what was
sent.
Examples
--------
>>> _describe_failure(TimeoutError("anything at all"))
'TimeoutError'
>>> import requests
>>> r = requests.Response(); r.status_code = 400
>>> _describe_failure(requests.HTTPError(response=r))
'HTTPError (HTTP 400)'
"""
text = type(exc).__name__
status = getattr(getattr(exc, "response", None), "status_code", None)
if isinstance(status, int):
text += f" (HTTP {status})"
return text
def warm_sources() -> tuple[str | None, ...]:
"""The data sources whose prefix a real request can start with.
``(None,)`` for a deployment with one data source, whose prompts are
built for the whole schema; the source names, default first, when
several are configured -- the same split
:func:`prompt_engine.source_scope.scoped_source` makes.
Examples
--------
>>> warm_sources()
(None,)
"""
import database.datasources as datasources
names = datasources.datasource_names()
return (None,) if len(names) < 2 else tuple(names)
def _prefix_segments(system_prompt: str, source: str | None) -> PromptSegments:
"""The cacheable part of a real prompt for *source*, and nothing else.
Built by :func:`~llm.router.build_prompt_segments` with an empty
question and context, so ``static_prefix`` is what real requests
carry; it is empty when *source* is on the retrieval path (no stable
prefix to warm).
Examples
--------
>>> seg = _prefix_segments("You are a T-SQL expert.", None)
>>> seg.static_prefix.startswith("\\nYou are a T-SQL expert.")
True
>>> seg.question, seg.session_context
('', '')
"""
from core.models import RetrievalContext
built = build_prompt_segments("", system_prompt, RetrievalContext(), source=source)
return PromptSegments(static_prefix=built.static_prefix)
class _OutcomeLog:
"""Says each source's outcome once per streak, then only at ``DEBUG``.
A keep-warm pass repeats every few minutes for as long as the server
runs; an endpoint that is down, or a source on the retrieval path, must
be one log line and not one per pass. The first time a source has an
outcome *kind* it is logged at the level the call site chose; the same
kind again is logged at ``DEBUG``; a different kind is a new streak and
is logged at its level again. A success that ends a streak of failures
says how many passes failed.
"""
def __init__(self) -> None:
self._lock = threading.Lock()
self._state: dict[str | None, tuple[str, int]] = {}
def failures_before(self, source: str | None) -> int:
"""How many consecutive failures *source* had up to now (0 if none)."""
with self._lock:
kind, count = self._state.get(source, ("", 0))
return count if kind == "failed" else 0
def emit(
self, source: str | None, kind: str, level: int, message: str, *args: object,
) -> None:
"""Log *message* at *level* if *kind* starts a streak for *source*, else at DEBUG."""
with self._lock:
previous, count = self._state.get(source, ("", 0))
new = previous != kind
self._state[source] = (kind, 1 if new else count + 1)
logger.log(level if new else logging.DEBUG, message, *args)
def _warm_one(
router: LLMRouter,
system_prompt: str,
source: str | None,
timeout: float,
*,
skip_used_within: float | None = None,
outcomes: _OutcomeLog | None = None,
activity: PrefixActivity | None = None,
) -> SourceWarmup:
"""Warm one source; never raises.
Parameters
----------
skip_used_within:
When given (the keep-warm), a source whose prefix a real request
used within this many seconds is skipped: that request already
refreshed the server's cache. ``None`` (the start-up pass and
``POST /admin/llm/warmup``) always sends.
outcomes:
When given (the keep-warm), repeated outcomes are logged once per
streak instead of every time.
activity:
The record of real requests; the process-wide one by default. A
successful warm-up is recorded in it too, so a keep-warm pass right
after a start-up or manual warm-up does not repeat it.
"""
label = _source_label(source)
activity = activity if activity is not None else _default_activity
def say(kind: str, level: int, message: str, *args: object) -> None:
if outcomes is not None:
outcomes.emit(source, kind, level, message, *args)
else:
logger.log(level, message, *args)
try:
segments = _prefix_segments(system_prompt, source)
except Exception as exc: # noqa: BLE001 - a warm-up must not fail anything else
say(
"skipped:unbuildable", logging.WARNING,
"LLM prefix warm-up: source=%s skipped, could not build the prefix (%s)",
label, _describe_failure(exc),
)
return SourceWarmup(source, "skipped", f"could not build the prefix ({_describe_failure(exc)})")
if not segments.static_prefix:
say(
"skipped:retrieval", logging.INFO,
"LLM prefix warm-up: source=%s skipped, its prompt uses retrieval, not a static prefix",
label,
)
return SourceWarmup(source, "skipped", "prompt uses retrieval, not a static prefix")
if skip_used_within is not None:
idle = activity.seconds_since_use(segments.static_prefix)
if idle is not None and idle < skip_used_within:
say(
"skipped:recent", logging.DEBUG,
"LLM prefix warm-up: source=%s skipped, a request used its prefix %.0f s ago",
label, idle,
)
return SourceWarmup(
source, "skipped", f"a request used this prefix {idle:.0f} s ago; it is warm",
)
start = time.perf_counter()
try:
report = router.warm_prefix(segments, timeout=timeout)
except NotImplementedError:
say(
"skipped:no-cache", logging.INFO,
"LLM prefix warm-up: source=%s skipped, the backend has no prefix cache", label,
)
return SourceWarmup(source, "skipped", "backend has no prefix cache to warm")
except RemoteProviderNotAllowedError:
say(
"skipped:remote", logging.INFO,
"LLM prefix warm-up: source=%s skipped, the endpoint is not trusted "
"and LLM_ALLOW_REMOTE is off", label,
)
return SourceWarmup(source, "skipped", "endpoint not allowed (LLM_ALLOW_REMOTE is off)")
except Exception as exc: # noqa: BLE001 - every failure is logged and swallowed
elapsed_ms = round((time.perf_counter() - start) * 1000)
what = _describe_failure(exc)
say(
"failed", logging.WARNING,
"LLM prefix warm-up: source=%s failed after %d ms (%s)", label, elapsed_ms, what,
)
return SourceWarmup(source, "failed", what, duration_ms=elapsed_ms)
elapsed_ms = round((time.perf_counter() - start) * 1000)
prompt_tokens = report.get("prompt_tokens")
cached_tokens = report.get("cached_tokens")
activity.note_use(segments.static_prefix)
recovered = outcomes.failures_before(source) if outcomes is not None else 0
say(
"warmed", logging.INFO,
"LLM prefix warm-up: source=%s warmed in %d ms, prompt_tokens=%s%s%s",
label, elapsed_ms,
prompt_tokens if prompt_tokens is not None else "unknown",
f", cached_tokens={cached_tokens}" if cached_tokens is not None else "",
f" (recovered after {recovered} failed pass(es))" if recovered else "",
)
return SourceWarmup(
source, "warmed", None, duration_ms=elapsed_ms,
prompt_tokens=prompt_tokens, cached_tokens=cached_tokens,
)
def warm_prefix_cache(
system_prompt: str,
router_factory: Callable[[], LLMRouter],
*,
timeout_seconds: float | None = None,
skip_used_within: float | None = None,
outcomes: _OutcomeLog | None = None,
activity: PrefixActivity | None = None,
cancel: Callable[[], bool] | None = None,
) -> list[SourceWarmup] | None:
"""Warm the prefix cache for every data source, within one time budget.
Sources are warmed one after another (a model server prefills them one
after another anyway); each request is given what is left of the
budget as its HTTP timeout, and a source reached after the budget is
spent is skipped. Never raises.
Parameters
----------
system_prompt:
The loaded system prompt, as ``api/server.py`` holds it.
router_factory:
Returns the :class:`~llm.router.LLMRouter` real requests use
(``api.runner``'s agent's router). Called inside this function so
that a failure to build it is logged rather than raised.
timeout_seconds:
Total budget in seconds; ``LLM_PREFIX_WARMUP_TIMEOUT_SECONDS`` when
``None``. The budget bounds each request's timeout, so a request
that connects slowly and then reads slowly can overrun it by up to
one more timeout; it is a bound, not a deadline to the millisecond.
skip_used_within, outcomes, activity:
What distinguishes a keep-warm pass (:class:`KeepWarm`) from the
start-up one; see :func:`_warm_one`. The defaults are the start-up
pass.
cancel:
Polled before each source; when it returns true the remaining
sources are skipped (the keep-warm's shutdown).
Returns
-------
list[SourceWarmup] | None
One entry per source, in order. ``None`` when another warm-up is
already running (nothing was sent).
Examples
--------
>>> from llm.providers import MockBackend
>>> router = LLMRouter(default_chain=[MockBackend()])
>>> [r.status for r in warm_prefix_cache("You are a T-SQL expert.", lambda: router)]
['skipped']
"""
if not _warmup_lock.acquire(blocking=False):
logger.log(
logging.DEBUG if outcomes is not None else logging.INFO,
"LLM prefix warm-up: another warm-up is already running, not starting a second",
)
return None
try:
budget = float(
timeout_seconds if timeout_seconds is not None
else cfg.settings.llm_prefix_warmup_timeout_seconds
)
deadline = time.perf_counter() + budget
try:
router = router_factory()
sources = warm_sources()
except Exception as exc: # noqa: BLE001 - nothing to warm, nothing to break
message = "LLM prefix warm-up: not started, could not set up (%s)"
if outcomes is not None:
outcomes.emit(None, "setup", logging.WARNING, message, _describe_failure(exc))
else:
logger.warning(message, _describe_failure(exc))
return [SourceWarmup(None, "skipped", f"could not set up ({_describe_failure(exc)})")]
results: list[SourceWarmup] = []
for source in sources:
if cancel is not None and cancel():
results.append(SourceWarmup(source, "skipped", "shutting down"))
continue
remaining = deadline - time.perf_counter()
if remaining <= 0:
logger.warning(
"LLM prefix warm-up: source=%s skipped, the %.0f s budget is spent",
_source_label(source), budget,
)
results.append(SourceWarmup(source, "skipped", "time budget exhausted"))
continue
results.append(_warm_one(
router, system_prompt, source, remaining,
skip_used_within=skip_used_within, outcomes=outcomes, activity=activity,
))
return results
finally:
_warmup_lock.release()
def start_background_warmup(
system_prompt: str,
router_factory: Callable[[], LLMRouter],
) -> threading.Thread:
"""Run :func:`warm_prefix_cache` on a daemon thread and return at once.
The thread is a daemon so that it can never keep the process alive at
shutdown, and it catches everything, so a bug in the warm-up is a log
line and not an unhandled-thread-exception traceback.
Parameters
----------
system_prompt, router_factory:
As for :func:`warm_prefix_cache`.
Returns
-------
threading.Thread
The started thread (tests ``join`` it; the server ignores it).
Examples
--------
>>> from llm.providers import MockBackend
>>> router = LLMRouter(default_chain=[MockBackend()])
>>> t = start_background_warmup("You are a T-SQL expert.", lambda: router)
>>> t.join(5); t.is_alive()
False
"""
def _run() -> None:
try:
warm_prefix_cache(system_prompt, router_factory)
except Exception as exc: # noqa: BLE001 - belt and braces: warm_prefix_cache already catches
logger.warning("LLM prefix warm-up: stopped unexpectedly (%s)", _describe_failure(exc))
thread = threading.Thread(target=_run, name="llm-prefix-warmup", daemon=True)
thread.start()
return thread
# ---------------------------------------------------------------------------
# Keeping the cache warm
# ---------------------------------------------------------------------------
class KeepWarm:
"""Repeats the warm-up every *interval_seconds*, skipping what real traffic keeps warm.
Each pass is :func:`warm_prefix_cache` (so the request, the
data-governance gate, the time budget and the "retrieval path means
nothing to warm" rule are the warm-up's own), with ``skip_used_within``
set to the interval. See the module docstring's "Keeping it warm".
The first pass runs one interval after :meth:`start`, not at once: the
start-up warm-up (or the first question) is what primes a cold server.
Parameters
----------
system_prompt, router_factory:
As for :func:`warm_prefix_cache`.
interval_seconds:
Seconds between passes, and the age under which a real request
counts as having kept a prefix warm.
paused:
Returns true while passes must not send anything (maintenance mode).
An exception from it counts as "not paused".
activity:
The record of real requests (:mod:`llm.prefix_activity`); the
process-wide one by default, which is the one the router writes to.
wait:
``wait(seconds) -> bool`` blocks for that long and returns true when
the keep-warm was stopped meanwhile. The default waits on the stop
event; tests pass a fake that advances a fake clock.
now:
Returns the current wall-clock time, for ``last_run_at``.
Examples
--------
>>> from llm.providers import MockBackend
>>> router = LLMRouter(default_chain=[MockBackend()])
>>> keepwarm = KeepWarm("You are a T-SQL expert.", lambda: router, 60)
>>> [r.status for r in keepwarm.run_once()]
['skipped']
>>> keepwarm.status()["last_run_outcome"]
'skipped'
"""
def __init__(
self,
system_prompt: str,
router_factory: Callable[[], LLMRouter],
interval_seconds: float,
*,
paused: Callable[[], bool] | None = None,
activity: PrefixActivity | None = None,
wait: Callable[[float], bool] | None = None,
now: Callable[[], datetime] | None = None,
) -> None:
self._system_prompt = system_prompt
self._router_factory = router_factory
self.interval_seconds = float(interval_seconds)
self._paused = paused
self._activity = activity if activity is not None else _default_activity
self._stop = threading.Event()
self._wait = wait if wait is not None else self._stop.wait
self._now = now if now is not None else (lambda: datetime.now(timezone.utc))
self._outcomes = _OutcomeLog()
self._thread: threading.Thread | None = None
self._lock = threading.Lock()
self._runs = 0
self._last_run_at: str | None = None
self._last_outcome: str | None = None
self._last_results: list[SourceWarmup] = []
# -- control ---------------------------------------------------------
def start(self) -> threading.Thread:
"""Start the daemon thread (once) and return it."""
with self._lock:
if self._thread is not None and self._thread.is_alive():
return self._thread
self._stop.clear()
self._thread = threading.Thread(
target=self._run, name="llm-prefix-keepwarm", daemon=True,
)
self._thread.start()
return self._thread
def stop(self, timeout: float = 5.0) -> bool:
"""Ask the thread to stop and wait up to *timeout* seconds for it.
Returns true when it has ended. A pass in the middle of a request
to the model server ends with that request (bounded by
``LLM_PREFIX_WARMUP_TIMEOUT_SECONDS``); the thread is a daemon, so a
process that is exiting does not wait for it.
"""
self._stop.set()
with self._lock:
thread = self._thread
if thread is not None and thread is not threading.current_thread():
thread.join(timeout)
return not thread.is_alive()
return True
@property
def running(self) -> bool:
"""True while the thread is alive and has not been asked to stop."""
thread = self._thread
return bool(thread is not None and thread.is_alive() and not self._stop.is_set())
# -- one pass --------------------------------------------------------
def _is_paused(self) -> bool:
if self._paused is None:
return False
try:
return bool(self._paused())
except Exception: # noqa: BLE001 - a broken pause check must not stop the keep-warm
return False
def run_once(self) -> list[SourceWarmup] | None:
"""Run one pass now; never raises.
Returns
-------
list[SourceWarmup] | None
One entry per source, as for :func:`warm_prefix_cache`; ``None``
when nothing was attempted (paused, stopped, or another warm-up
was running).
"""
if self._stop.is_set():
return None
if self._is_paused():
self._record("paused", [])
return None
try:
results = warm_prefix_cache(
self._system_prompt, self._router_factory,
skip_used_within=self.interval_seconds,
outcomes=self._outcomes, activity=self._activity,
cancel=self._stop.is_set,
)
except Exception as exc: # noqa: BLE001 - belt and braces: warm_prefix_cache already catches
self._outcomes.emit(
None, "crashed", logging.WARNING,
"LLM prefix keep-warm: a pass stopped unexpectedly (%s)", _describe_failure(exc),
)
self._record("failed", [SourceWarmup(None, "failed", _describe_failure(exc))])
return None
if results is None:
self._record("busy", [])
return None
if any(r.status == "failed" for r in results):
outcome = "failed"
elif any(r.status == "warmed" for r in results):
outcome = "warmed"
else:
outcome = "skipped"
self._record(outcome, results)
return results
def _record(self, outcome: str, results: list[SourceWarmup]) -> None:
with self._lock:
self._runs += 1
self._last_run_at = self._now().isoformat()
self._last_outcome = outcome
self._last_results = list(results)
def _run(self) -> None:
logger.info(
"LLM prefix keep-warm: every %.0f s; a source a real request used within that "
"time is skipped", self.interval_seconds,
)
while not self._stop.is_set():
if self._wait(self.interval_seconds):
break
self.run_once()
logger.info("LLM prefix keep-warm: stopped")
# -- what the admin routes show ---------------------------------------
def status(self) -> dict[str, Any]:
"""A JSON-ready account of the keep-warm and its last pass.
``last_run_outcome`` is ``"warmed"`` (at least one prefix was sent
and answered), ``"skipped"`` (every source was skipped: kept warm by
traffic, on the retrieval path, or not allowed), ``"failed"`` (at
least one request failed), ``"paused"`` (maintenance mode) or
``"busy"`` (another warm-up was running); ``None`` before the first
pass. ``last_run`` has one entry per source, as
``POST /admin/llm/warmup`` does. No prompt text anywhere.
"""
paused = self._is_paused()
with self._lock:
return {
"enabled": True,
"interval_seconds": self.interval_seconds,
"running": self.running,
"paused": paused,
"runs": self._runs,
"last_run_at": self._last_run_at,
"last_run_outcome": self._last_outcome,
"last_run": [r.as_dict() for r in self._last_results],
}
#: The keep-warm :func:`start_keepwarm` started, if any.
_keepwarm: KeepWarm | None = None
_keepwarm_lock = threading.Lock()
def start_keepwarm(
system_prompt: str,
router_factory: Callable[[], LLMRouter],
*,
paused: Callable[[], bool] | None = None,
) -> KeepWarm | None:
"""Start the process's keep-warm if ``LLM_PREFIX_KEEPWARM_SECONDS`` asks for one.
Returns ``None`` (and starts nothing) when the setting is ``0``, when
``LLM_PROVIDER=mock`` (there is no model server to keep warm), or when a
keep-warm is already running. Never raises.
Examples
--------
>>> start_keepwarm("You are a T-SQL expert.", lambda: None) is None
True
"""
global _keepwarm
try:
seconds = float(cfg.settings.llm_prefix_keepwarm_seconds)
if seconds <= 0:
return None
if str(cfg.settings.llm_provider).lower() == "mock":
logger.info("LLM prefix keep-warm: not started, LLM_PROVIDER=mock has no model server")
return None
with _keepwarm_lock:
if _keepwarm is not None and _keepwarm.running:
return None
keepwarm = KeepWarm(system_prompt, router_factory, seconds, paused=paused)
keepwarm.start()
_keepwarm = keepwarm
return keepwarm
except Exception as exc: # noqa: BLE001 - an optimisation, startup must not fail on it
logger.warning("LLM prefix keep-warm could not be started: %s", _describe_failure(exc))
return None
def stop_keepwarm(timeout: float = 5.0) -> None:
"""Stop the process's keep-warm, if one is running (``lifespan`` shutdown). Never raises."""
global _keepwarm
with _keepwarm_lock:
keepwarm, _keepwarm = _keepwarm, None
if keepwarm is None:
return
try:
if not keepwarm.stop(timeout):
logger.warning(
"LLM prefix keep-warm: still finishing a request to the model server at "
"shutdown; it is a daemon thread and ends with the process",
)
except Exception as exc: # noqa: BLE001 - shutdown must not fail on it
logger.warning("LLM prefix keep-warm could not be stopped: %s", _describe_failure(exc))
def keepwarm_status() -> dict[str, Any]:
"""The keep-warm's status, or the reason there is none.
With no keep-warm running: ``{"enabled": False, "reason": ...}`` and the
same keys with empty values. Safe to call from a request.
Examples
--------
>>> keepwarm_status()["enabled"]
False
"""
with _keepwarm_lock:
keepwarm = _keepwarm
if keepwarm is not None:
return keepwarm.status()
seconds = float(cfg.settings.llm_prefix_keepwarm_seconds)
if seconds <= 0:
reason = "LLM_PREFIX_KEEPWARM_SECONDS is 0"
elif str(cfg.settings.llm_provider).lower() == "mock":
reason = "LLM_PROVIDER=mock has no model server"
else:
reason = "not started"
return {
"enabled": False,
"reason": reason,
"interval_seconds": seconds,
"running": False,
"paused": False,
"runs": 0,
"last_run_at": None,
"last_run_outcome": None,
"last_run": [],
}