from __future__ import annotations from .shared import * class LLMCacheCircuitMixin: _provider_circuit_state: dict[str, dict[str, Any]] = {} _provider_circuit_config_signature: str = "" def __init__(self) -> None: self.request_timeout = float(os.getenv("LLM_REQUEST_TIMEOUT", "45")) self.redis_url = os.getenv("REDIS_URL") self._redis: Optional[Any] = None self._prompt_profile_cache: dict[int, tuple[float, dict[str, str]]] = {} self._openclaw_routing_cache: tuple[float, dict[str, Any]] | None = None type(self)._provider_circuit_config_signature = type(self)._current_model_config_signature() async def _get_redis(self): if not REDIS_AVAILABLE or not self.redis_url: return None if self._redis is None: self._redis = redis_async.from_url(self.redis_url, decode_responses=True) return self._redis @staticmethod def _normalize_prompt_for_hash(prompt: str) -> str: return re.sub(r"\s+", " ", str(prompt or "")).strip() @classmethod def _structured_cache_key(cls, *, function_name: str, model: str, prompt: str) -> str: normalized_prompt = cls._normalize_prompt_for_hash(prompt) digest = hashlib.sha256(normalized_prompt.encode("utf-8")).hexdigest() return f"pmflow:structured:{function_name}:{model}:{digest}" async def _structured_cache_get(self, *, key: str) -> Optional[dict[str, Any]]: redis = await self._get_redis() if redis is None: return None try: raw = await redis.get(key) except Exception: return None if not raw: return None try: parsed = json.loads(raw) except Exception: return None return parsed if isinstance(parsed, dict) else None async def _structured_cache_set(self, *, key: str, value: dict[str, Any], ttl: int = STRUCTURED_CACHE_TTL_SECONDS) -> None: redis = await self._get_redis() if redis is None: return try: await redis.setex(key, ttl, json.dumps(value, ensure_ascii=False)) except Exception: return @staticmethod def _provider_circuit_key(*, provider: str, base_url: str, model: str) -> str: return f"{str(provider or '').strip().lower()}|{str(base_url or '').strip().rstrip('/')}|{str(model or '').strip()}" @classmethod def _current_model_config_signature(cls) -> str: bundle = load_model_config() config_path = str(bundle.get("config_path") or "").strip() if not config_path: return "" try: mtime = Path(config_path).expanduser().stat().st_mtime except Exception: mtime = -1.0 version = str(bundle.get("version") or "").strip() updated_at = str(bundle.get("updated_at") or "").strip() return f"{config_path}|{mtime}|{version}|{updated_at}" @classmethod def _refresh_provider_circuit_state_if_needed(cls) -> None: current_signature = cls._current_model_config_signature() if not current_signature: return previous_signature = str(cls._provider_circuit_config_signature or "").strip() if previous_signature and previous_signature != current_signature: cls._provider_circuit_state.clear() cls._provider_circuit_config_signature = current_signature @staticmethod def _http_status_from_exception(exc: Exception) -> Optional[int]: if isinstance(exc, httpx.HTTPStatusError) and exc.response is not None: try: return int(exc.response.status_code) except Exception: return None return None @staticmethod def _is_retryable_http_status(status_code: Optional[int]) -> bool: if status_code is None: return False return int(status_code) in {408, 409, 429, 500, 502, 503, 504} @staticmethod def _parse_retry_after_seconds(raw_value: Optional[str]) -> Optional[float]: value = str(raw_value or "").strip() if not value: return None try: seconds = float(value) except Exception: return None if seconds <= 0: return None return min(seconds, 10.0) @classmethod def _http_retry_delay_seconds( cls, *, status_code: Optional[int], attempt: int, retry_after_header: Optional[str], ) -> float: retry_after = cls._parse_retry_after_seconds(retry_after_header) if retry_after is not None: return retry_after base = 0.8 if int(status_code or 0) == 429: base = 1.2 capped_attempt = max(0, min(int(attempt), 6)) return min(base * (2 ** capped_attempt), 6.0) @classmethod def _is_fail_fast_exception(cls, exc: Exception) -> bool: status_code = cls._http_status_from_exception(exc) if status_code in {401, 403}: return True text = str(exc or "").strip().lower() return any(marker in text for marker in FAIL_FAST_ERROR_MARKERS) @classmethod def _is_provider_circuit_open_exception(cls, exc: Exception) -> bool: return "provider_circuit_open" in str(exc or "").strip().lower() @classmethod def _is_circuit_breaker_worthy_exception(cls, exc: Exception) -> bool: if cls._is_fail_fast_exception(exc): return True if isinstance(exc, httpx.RequestError): return True status_code = cls._http_status_from_exception(exc) return cls._is_retryable_http_status(status_code) @classmethod def _is_provider_circuit_open(cls, circuit_key: str) -> bool: cls._refresh_provider_circuit_state_if_needed() state = cls._provider_circuit_state.get(str(circuit_key)) if not state: return False try: open_until = float(state.get("open_until") or 0.0) except Exception: open_until = 0.0 now = time.monotonic() if open_until <= now: if open_until: cls._provider_circuit_state.pop(str(circuit_key), None) return False return True @classmethod def _record_provider_success(cls, circuit_key: str) -> None: cls._refresh_provider_circuit_state_if_needed() cls._provider_circuit_state.pop(str(circuit_key), None) @classmethod def _record_provider_failure(cls, circuit_key: str, exc: Exception) -> None: cls._refresh_provider_circuit_state_if_needed() if cls._is_provider_circuit_open_exception(exc): return if not cls._is_circuit_breaker_worthy_exception(exc): return now = time.monotonic() state = dict(cls._provider_circuit_state.get(str(circuit_key)) or {}) consecutive = int(state.get("consecutive_failures") or 0) + 1 open_until = float(state.get("open_until") or 0.0) if consecutive >= CIRCUIT_BREAKER_FAILURE_THRESHOLD: open_until = now + CIRCUIT_BREAKER_COOLDOWN_SECONDS cls._provider_circuit_state[str(circuit_key)] = { "consecutive_failures": consecutive, "open_until": open_until, "last_error": str(exc)[:240], "updated_at_monotonic": now, }