import logging import time from ..exceptions import LLMGenerationError from .base import LLMProvider logger = logging.getLogger(__name__) # Process-level: provider names added here are skipped on all future calls until restart. _exhausted: set[str] = set() # Tracks when each rate-limited provider should be retryable again (unix timestamp). _rate_limited_until: dict[str, float] = {} def _is_credit_exhausted(exc: Exception) -> bool: """Returns True for unrecoverable billing/quota errors that should permanently blacklist a provider.""" try: import openai if isinstance(exc, openai.AuthenticationError): return True if isinstance(exc, openai.RateLimitError): code = getattr(exc, "code", "") or "" if code in ("insufficient_quota", "quota_exceeded"): return True msg = str(exc).lower() return "quota" in msg or "exceeded your current" in msg or "billing" in msg except ImportError: pass try: import anthropic if isinstance(exc, anthropic.AuthenticationError): return True if isinstance(exc, (anthropic.RateLimitError, anthropic.BadRequestError)): msg = str(exc).lower() return "credit" in msg or "balance" in msg or "too low" in msg except ImportError: pass return False def _is_rate_limited(exc: Exception) -> bool: """Returns True for transient 429 rate-limit errors (retry next provider, don't blacklist).""" try: import openai if isinstance(exc, openai.RateLimitError): code = getattr(exc, "code", "") or "" if code not in ("insufficient_quota", "quota_exceeded"): msg = str(exc).lower() if "quota" not in msg and "billing" not in msg and "exceeded your current" not in msg: return True except ImportError: pass try: import anthropic if isinstance(exc, anthropic.RateLimitError): msg = str(exc).lower() if "credit" not in msg and "balance" not in msg and "too low" not in msg: return True except ImportError: pass status = getattr(exc, "status_code", None) or getattr(exc, "status", None) if status == 429: return not _is_credit_exhausted(exc) return False def _extract_retry_after(exc: Exception) -> int: """Extract how many seconds until this provider's rate limit resets. Falls back to 60s (typical for Gemini/OpenAI free tier per-minute windows). """ response = getattr(exc, "response", None) if response is not None: headers = getattr(response, "headers", {}) or {} raw = headers.get("retry-after") or headers.get("Retry-After") if raw: try: return max(1, int(raw)) except (ValueError, TypeError): pass return 60 class FallbackProvider(LLMProvider): """Tries providers in priority order. - Credit/quota exhausted → permanently blacklist the provider for this process lifetime. - Rate limited (429, transient) → skip to next provider without blacklisting; records when the rate-limited provider should be available again. - Any other error → re-raise immediately (don't try fallbacks for unexpected failures). After a successful call, `last_used` and `skipped` expose which provider handled the request and which ones were bypassed (with retry-after info), so callers can surface this to the user. """ def __init__(self, providers: list[tuple[str, LLMProvider]]) -> None: self._providers = providers # Set after each complete_json call self._last_used: str | None = None self._skipped: list[dict] = [] # [{name, reason, retry_after_seconds}] # ── per-call result metadata ────────────────────────────────────────────── @property def last_used(self) -> str | None: return self._last_used @property def skipped(self) -> list[dict]: return self._skipped # ── public helpers used by /providers/status ────────────────────────────── @property def names(self) -> list[str]: return [n for n, _ in self._providers] @property def active(self) -> list[str]: now = time.time() return [n for n, _ in self._providers if n not in _exhausted and _rate_limited_until.get(n, 0) <= now] @property def exhausted(self) -> list[str]: return [n for n in _exhausted if any(n == name for name, _ in self._providers)] def reset(self, name: str | None = None) -> None: """Clear exhausted/rate-limited state (all providers, or just one by name).""" global _exhausted, _rate_limited_until if name: _exhausted.discard(name) _rate_limited_until.pop(name, None) else: _exhausted.clear() _rate_limited_until.clear() # ── LLMProvider interface ───────────────────────────────────────────────── def complete_json(self, system_prompt: str, messages: list[dict]) -> dict: self._last_used = None self._skipped = [] last_exc: Exception | None = None now = time.time() for name, provider in self._providers: if name in _exhausted: logger.debug("Skipping exhausted provider '%s'", name) self._skipped.append({"name": name, "reason": "credit_exhausted", "retry_after_seconds": None}) continue if _rate_limited_until.get(name, 0) > now: secs_left = int(_rate_limited_until[name] - now) logger.debug("Skipping rate-limited provider '%s' (%ds left)", name, secs_left) self._skipped.append({"name": name, "reason": "rate_limited", "retry_after_seconds": secs_left}) continue try: result = provider.complete_json(system_prompt, messages) self._last_used = name if self._skipped: logger.info("LLM call succeeded via '%s' (skipped: %s)", name, [s["name"] for s in self._skipped]) return result except Exception as exc: if _is_credit_exhausted(exc): logger.warning("Provider '%s' out of credit — blacklisting. Error: %s", name, exc) _exhausted.add(name) self._skipped.append({"name": name, "reason": "credit_exhausted", "retry_after_seconds": None}) last_exc = exc elif _is_rate_limited(exc): retry_after = _extract_retry_after(exc) _rate_limited_until[name] = now + retry_after logger.warning("Provider '%s' rate-limited for %ds — trying fallback.", name, retry_after) self._skipped.append({"name": name, "reason": "rate_limited", "retry_after_seconds": retry_after}) last_exc = exc else: raise skipped_names = [s["name"] for s in self._skipped] raise LLMGenerationError( "All LLM providers are unavailable.", detail=( f"Providers tried: {', '.join(skipped_names)}. " f"Last error: {last_exc}. " "Add credit to a provider or wait for rate limits to reset." ), )