""" llm_cache.py — Gap 2.3: Distributed LLM response cache via Upstash Redis REST. Architettura: - Upstash Redis REST API (HTTP, no TCP persistente — funziona su HF Spaces/Railway) - Cache key: sha256(model + json(messages))[:32], prefisso "llm:" - TTL: 3600s (1 ora, configurabile via LLM_CACHE_TTL) - Graceful degradation: no-op completo se UPSTASH_REDIS_REST_URL non impostato - Cache solo chat() non-streaming con temperature ≤ 0.5 e senza tool results - Max value size: 32 KB — skip write se risposta più lunga - Stats in-process: hits/misses/writes/errors/skipped + hit_rate Env vars: UPSTASH_REDIS_REST_URL — es. https://xxx.upstash.io (obbligatoria per attivare) UPSTASH_REDIS_REST_TOKEN — Bearer token Upstash LLM_CACHE_TTL — secondi TTL per entry (default: 3600) LLM_CACHE_MAX_BYTES — max byte per value (default: 32768 = 32 KB) """ from __future__ import annotations import hashlib import json import logging import os import httpx from fastapi import APIRouter router = APIRouter(prefix="/api/cache", tags=["cache"]) _logger = logging.getLogger("llm_cache") _CACHE_TTL_S = int(os.getenv("LLM_CACHE_TTL", "3600")) _CACHE_MAX_BYTES = int(os.getenv("LLM_CACHE_MAX_BYTES", "32768")) # 32 KB _KEY_PREFIX = "llm:" # ── Stats in-process (reset a ogni restart) ─────────────────────────────────── _stats: dict[str, int] = { "hits": 0, "misses": 0, "writes": 0, "errors": 0, "skipped": 0, } def _cfg() -> tuple[str, str] | tuple[None, None]: """Ritorna (url, token) Upstash o (None, None) se non configurato.""" url = (os.getenv("UPSTASH_REDIS_REST_URL") or "").rstrip("/") token = (os.getenv("UPSTASH_REDIS_REST_TOKEN") or "").strip() return (url, token) if (url and token) else (None, None) # ── Cache key ────────────────────────────────────────────────────────────────── def cache_key(model: str, messages: list) -> str: """SHA-256 deterministico dei messaggi — chiave identica tra istanze diverse.""" payload = json.dumps( {"m": model, "msgs": messages}, sort_keys=True, ensure_ascii=True, separators=(",", ":"), ) return _KEY_PREFIX + hashlib.sha256(payload.encode()).hexdigest()[:32] def should_cache(messages: list, temperature: float) -> bool: """ True se la coppia (messages, temperature) è cacheabile in modo sicuro. Skip quando: - temperature > 0.5 (output non-deterministico — alta varianza) - almeno 1 message con role "tool" (risultati specifici del contesto agente) - system prompt > 8000 chars (variabilità troppo alta, hit-rate bassa) """ if temperature > 0.5: return False for m in messages: role = m.get("role", "") if role == "tool": return False if role == "system" and len(str(m.get("content") or "")) > 8000: return False return True # ── Upstash Redis REST GET ──────────────────────────────────────────────────── async def get_cached(key: str) -> str | None: """ GET da Upstash Redis REST. Timeout 2s — non blocca mai il path principale LLM. Ritorna None su miss, errore o cache non configurata. """ url, token = _cfg() if not url: return None try: async with httpx.AsyncClient(timeout=2.0) as c: r = await c.get( f"{url}/get/{key}", headers={"Authorization": f"Bearer {token}"}, ) if r.status_code != 200: _stats["errors"] += 1 return None result = r.json().get("result") if result is not None: _stats["hits"] += 1 _logger.debug("llm_cache HIT key=…%s", key[-8:]) return str(result) _stats["misses"] += 1 _logger.debug("llm_cache MISS key=…%s", key[-8:]) return None except Exception as _e: _stats["errors"] += 1 _logger.debug("llm_cache GET error (%s)", type(_e).__name__) return None # ── Upstash Redis REST SET ──────────────────────────────────────────────────── async def set_cached(key: str, value: str, ttl_s: int = _CACHE_TTL_S) -> None: """ SET su Upstash Redis REST via pipeline endpoint. Fire-and-forget — non rilancia mai eccezioni. Skippa se il value supera _CACHE_MAX_BYTES (risposta molto lunga, hit-rate bassa). """ _value_bytes = len(value.encode("utf-8")) if _value_bytes > _CACHE_MAX_BYTES: _stats["skipped"] += 1 _logger.debug( "llm_cache SKIP write key=…%s (size=%d B > max=%d B)", key[-8:], _value_bytes, _CACHE_MAX_BYTES, ) return url, token = _cfg() if not url: return try: async with httpx.AsyncClient(timeout=3.0) as c: r = await c.post( f"{url}/pipeline", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, content=json.dumps([["SET", key, value, "EX", str(ttl_s)]]).encode(), ) if r.status_code == 200: _stats["writes"] += 1 _logger.debug("llm_cache WRITE key=…%s ttl=%ds", key[-8:], ttl_s) else: _stats["errors"] += 1 _logger.warning("llm_cache write HTTP %d: %s", r.status_code, r.text[:80]) except Exception as _e: _stats["errors"] += 1 _logger.debug("llm_cache SET error (%s)", type(_e).__name__) def get_stats() -> dict: url, _ = _cfg() total = _stats["hits"] + _stats["misses"] out: dict = { "configured": url is not None, "backend": "upstash_redis_rest" if url else "disabled", "ttl_s": _CACHE_TTL_S, "max_value_bytes": _CACHE_MAX_BYTES, **_stats, "hit_rate": round(_stats["hits"] / total, 3) if total else 0.0, } if not url: out["note"] = ( "Impostare UPSTASH_REDIS_REST_URL e UPSTASH_REDIS_REST_TOKEN " "in Railway/HF Spaces per attivare il caching distribuito LLM." ) return out # ── Router ──────────────────────────────────────────────────────────────────── @router.get("/stats") async def cache_stats(): """ Gap 2.3 observability: statistiche cache LLM distribuita. Ritorna: - configured: True se Upstash è configurato via env vars - backend: "upstash_redis_rest" | "disabled" - hits / misses / writes / errors / skipped (dall'ultimo restart) - hit_rate: hits / (hits + misses) - ttl_s, max_value_bytes: parametri di configurazione - note: istruzioni setup se non configurato """ return get_stats()