perf(auto-router): cache quality-signal responses in-process for an hour

The signals are directional evidence over a multi-day window; minute-level
freshness buys nothing, and each miss is a capped 100k-row spend-log scan.
Responses are now cached per (start_date, end_date) in an InMemoryCache with
a 1-hour TTL, so repeated dashboard loads stop re-scanning the table. Rejected
oversized windows are never cached -- narrowing the range must take effect
immediately.

Co-Authored-By: Claude <noreply@anthropic.com>

litellm/proxy/management_endpoints/auto_router_endpoints.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Annotated, Final
 from pydantic import BaseModel, TypeAdapter
 
 from litellm._logging import verbose_proxy_logger
+from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.exceptions import BudgetExceededError
 from litellm.proxy._types import (
     CommonProxyErrors,
@@ -498,6 +499,12 @@ _QUALITY_TURN_ROWS: Final = TypeAdapter(list[_QualityTurnRow])
 
 MAX_QUALITY_SIGNAL_ROWS: Final = 100_000
 
+QUALITY_SIGNALS_CACHE_TTL_SECONDS: Final = 3600
+
+_quality_signals_cache: Final = InMemoryCache(
+    max_size_in_memory=64, default_ttl=QUALITY_SIGNALS_CACHE_TTL_SECONDS
+)
+
 _QUALITY_SIGNALS_SQL: Final = """
 SELECT
     session_id,
@@ -652,6 +659,11 @@ async def get_auto_router_quality_signals(
     if end_day < start_day:
         raise HTTPException(status_code=400, detail="end_date must not be earlier than start_date")
 
+    cache_key: Final = f"{start_day.date()}:{end_day.date()}"
+    cached: Final = _quality_signals_cache.get_cache(cache_key)
+    if cached is not None:
+        return cached
+
     raw_rows: Final = await prisma_client.db.query_raw(
         _QUALITY_SIGNALS_SQL,
         start_day.isoformat(),
@@ -669,9 +681,11 @@ async def get_auto_router_quality_signals(
         )
     turns: Final = _QUALITY_TURN_ROWS.validate_python(raw_rows or ())
     router_names: Final = tuple(sorted(frozenset(row.router_name for row in turns if row.router_name is not None)))
-    return AutoRouterQualitySignalsResponse(
+    response: Final = AutoRouterQualitySignalsResponse(
         start_date=start_day.strftime("%Y-%m-%d"),
         end_date=end_day.strftime("%Y-%m-%d"),
         totals=_quality_signals_for(turns, None, llm_router),
         groups=tuple(_quality_signals_for(turns, name, llm_router) for name in router_names),
     )
+    _quality_signals_cache.set_cache(cache_key, response)
+    return response

tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py

@@ -477,6 +477,16 @@ class TestAutoRouterQualitySignals:
     CHEAP = "openai/gpt-4o-mini"
     PRICEY = "openai/gpt-4o"
 
+    @pytest.fixture(autouse=True)
+    def _fresh_cache(self):
+        from litellm.proxy.management_endpoints.auto_router_endpoints import (
+            _quality_signals_cache,
+        )
+
+        _quality_signals_cache.cache_dict.clear()
+        _quality_signals_cache.ttl_dict.clear()
+        yield
+
     @staticmethod
     def _row(
         session_id: str,
@@ -709,3 +719,66 @@ class TestAutoRouterQualitySignals:
         response = await self._call(rows, monkeypatch)
         assert response.totals.routed.sessions == 1
         assert response.totals.routed.abandonment_rate_pct == 0.0
+
+    @pytest.mark.asyncio
+    async def test_a_repeated_window_is_served_from_cache_without_a_second_scan(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        from litellm.proxy import proxy_server
+        from litellm.proxy.management_endpoints.auto_router_endpoints import (
+            get_auto_router_quality_signals,
+        )
+
+        scans = 0
+        rows = [self._row("s1", self.CHEAP, 1.0), self._row("s1", self.PRICEY, 2.0)]
+
+        class _DB:
+            async def query_raw(self, sql: str, *params: object):
+                nonlocal scans
+                scans += 1
+                return rows
+
+        monkeypatch.setattr(proxy_server, "prisma_client", type("P", (), {"db": _DB()})())
+        monkeypatch.setattr(proxy_server, "llm_router", self._pricing_router())
+
+        first = await get_auto_router_quality_signals(
+            user_api_key_dict=ADMIN, start_date="2026-08-01", end_date="2026-08-02"
+        )
+        second = await get_auto_router_quality_signals(
+            user_api_key_dict=ADMIN, start_date="2026-08-01", end_date="2026-08-02"
+        )
+        assert scans == 1
+        assert second == first
+
+        await get_auto_router_quality_signals(
+            user_api_key_dict=ADMIN, start_date="2026-08-01", end_date="2026-08-03"
+        )
+        assert scans == 2
+
+    @pytest.mark.asyncio
+    async def test_a_window_too_large_to_scan_is_not_cached_as_an_answer(self, monkeypatch: pytest.MonkeyPatch):
+        from litellm.proxy import proxy_server
+        from litellm.proxy.management_endpoints.auto_router_endpoints import (
+            MAX_QUALITY_SIGNAL_ROWS,
+            get_auto_router_quality_signals,
+        )
+
+        rows = [self._row(f"s{i}", self.CHEAP, 1.0) for i in range(MAX_QUALITY_SIGNAL_ROWS + 1)]
+
+        class _DB:
+            async def query_raw(self, sql: str, *params: object):
+                return rows
+
+        monkeypatch.setattr(proxy_server, "prisma_client", type("P", (), {"db": _DB()})())
+        monkeypatch.setattr(proxy_server, "llm_router", self._pricing_router())
+
+        with pytest.raises(HTTPException) as err:
+            await get_auto_router_quality_signals(
+                user_api_key_dict=ADMIN, start_date="2026-08-01", end_date="2026-08-02"
+            )
+        assert err.value.status_code == 400
+
+        with pytest.raises(HTTPException):
+            await get_auto_router_quality_signals(
+                user_api_key_dict=ADMIN, start_date="2026-08-01", end_date="2026-08-02"
+            )