""" Multi-provider LLM client with cascading fallback. Adopts pattern from Enterprise-AI-Gateway for resilient LLM access. """ import os import time import requests from typing import Optional, Tuple # Retry configuration - rotate through providers instead of consecutive retries MAX_ROUNDS = 3 # Number of times to cycle through all providers PROVIDER_DELAY = 10 # seconds between provider attempts class LLMClient: """LLM client with automatic provider fallback.""" def __init__(self, override_keys: dict = None): """Initialize client with available providers based on API keys. Args: override_keys: Optional dict with user-provided API keys. Keys: "groq", "gemini", "openrouter" """ self.providers = [] override_keys = override_keys or {} # Build providers list - use override keys if provided, else env vars groq_key = override_keys.get("groq") or os.getenv("GROQ_API_KEY") gemini_key = override_keys.get("gemini") or os.getenv("GEMINI_API_KEY") openrouter_key = override_keys.get("openrouter") or os.getenv("OPENROUTER_API_KEY") if groq_key: self.providers.append({ "name": "groq", "key": groq_key, "model": os.getenv("GROQ_MODEL", "llama-3.1-8b-instant"), "url": "https://api.groq.com/openai/v1/chat/completions" }) if gemini_key: self.providers.append({ "name": "gemini", "key": gemini_key, "model": os.getenv("GEMINI_MODEL", "gemini-2.0-flash") }) if openrouter_key: self.providers.append({ "name": "openrouter", "key": openrouter_key, "model": os.getenv("OPENROUTER_MODEL", "google/gemma-3n-e4b-it:free"), "url": "https://openrouter.ai/api/v1/chat/completions" }) if not self.providers: raise ValueError("No LLM API keys configured. Set at least one of: GROQ_API_KEY, GEMINI_API_KEY, OPENROUTER_API_KEY") def query(self, prompt: str, temperature: float = 0, max_tokens: int = 2048) -> Tuple[Optional[str], Optional[str], Optional[str], list]: """ Query LLM with rotating fallback across providers. Instead of retrying same provider consecutively, rotates: Groq → Gemini → OpenRouter → Groq → Gemini → OpenRouter → ... Returns: Tuple of (response_content, provider_used, error_message, providers_failed) providers_failed is a list of dicts: [{"name": "gemini", "error": "..."}] """ errors = [] providers_failed = [] is_first_attempt = True # Rotate through providers for MAX_ROUNDS cycles for round_num in range(MAX_ROUNDS): for provider in self.providers: # Add delay between attempts (skip first attempt) if not is_first_attempt: print(f"Waiting {PROVIDER_DELAY}s before trying {provider['name']} (round {round_num + 1})...") time.sleep(PROVIDER_DELAY) is_first_attempt = False print(f"Attempting LLM call with {provider['name']} (round {round_num + 1}/{MAX_ROUNDS})...") start_time = time.perf_counter() try: content, error = self._call_provider( provider=provider, prompt=prompt, temperature=temperature, max_tokens=max_tokens ) latency_ms = int((time.perf_counter() - start_time) * 1000) if content: print(f"Success with {provider['name']} ({latency_ms}ms)") provider_info = f"{provider['name']}:{provider['model']}" return content, provider_info, None, providers_failed else: errors.append(f"{provider['name']}: {error}") providers_failed.append({"name": provider['name'], "error": error}) print(f"Provider {provider['name']} failed: {error}") except Exception as e: errors.append(f"{provider['name']}: {str(e)}") providers_failed.append({"name": provider['name'], "error": str(e)}) print(f"Provider {provider['name']} exception: {e}") return None, None, f"All LLM providers failed after {MAX_ROUNDS} rounds: {'; '.join(errors)}", providers_failed def _make_request(self, url: str, headers: dict, payload: dict, provider_name: str) -> requests.Response: """Make HTTP request to provider (no internal retry - rotation handles retries).""" response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() return response def _call_provider(self, provider: dict, prompt: str, temperature: float, max_tokens: int) -> Tuple[Optional[str], Optional[str]]: """Call a specific LLM provider.""" headers = {"Content-Type": "application/json"} if provider["name"] == "groq": headers["Authorization"] = f"Bearer {provider['key']}" payload = { "model": provider["model"], "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature, } response = self._make_request(provider["url"], headers, payload, provider["name"]) data = response.json() if data and "choices" in data and data["choices"]: return data["choices"][0]["message"]["content"], None return None, "No content in Groq response" elif provider["name"] == "gemini": url = f"https://generativelanguage.googleapis.com/v1beta/models/{provider['model']}:generateContent?key={provider['key']}" payload = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens, } } response = self._make_request(url, headers, payload, provider["name"]) data = response.json() if data and "candidates" in data and data["candidates"]: first_candidate = data["candidates"][0] if "content" in first_candidate and "parts" in first_candidate["content"]: for part in first_candidate["content"]["parts"]: if "text" in part: return part["text"], None return None, "No text content in Gemini response" elif provider["name"] == "openrouter": headers["Authorization"] = f"Bearer {provider['key']}" headers["HTTP-Referer"] = "https://huggingface.co/spaces" headers["X-Title"] = "Instant SWOT Agent" payload = { "model": provider["model"], "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature, } response = self._make_request(provider["url"], headers, payload, provider["name"]) data = response.json() if data and "choices" in data and data["choices"]: return data["choices"][0]["message"]["content"], None return None, "No content in OpenRouter response" return None, f"Unknown provider: {provider['name']}" # Singleton instance for default (env-based) client _client = None def get_llm_client(override_keys: dict = None) -> LLMClient: """Get or create an LLM client instance. Args: override_keys: If provided, creates a new client with these keys. If None/empty, returns the singleton instance. """ # If user provided override keys, create a fresh client for this request if override_keys: return LLMClient(override_keys) # Otherwise use singleton for default env-based keys global _client if _client is None: _client = LLMClient() return _client