""" Multi-LLM abstraction layer. Supports Claude (Anthropic), OpenAI, Gemini, OpenRouter, and any OpenAI-compatible custom endpoint via direct HTTP calls using httpx. """ from abc import ABC, abstractmethod import httpx # Default models per provider DEFAULT_MODELS = { "claude": "claude-sonnet-4-20250514", "openai": "gpt-4o-mini", "gemini": "gemini-2.0-flash", "openrouter": "openai/gpt-4o-mini", "custom": "", } TIMEOUT = 60.0 class LLMProvider(ABC): """Abstract base class for LLM providers.""" def __init__(self, api_key: str, model: str | None = None): self.api_key = api_key self.model = model @abstractmethod def generate(self, prompt: str, system: str = "") -> str: """Generate text from a prompt.""" ... class ClaudeProvider(LLMProvider): """Anthropic Claude provider via Messages API.""" API_URL = "https://api.anthropic.com/v1/messages" def __init__(self, api_key: str, model: str | None = None): super().__init__(api_key, model or DEFAULT_MODELS["claude"]) def generate(self, prompt: str, system: str = "") -> str: headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", } payload: dict = { "model": self.model, "max_tokens": 2048, "messages": [{"role": "user", "content": prompt}], } if system: payload["system"] = system try: with httpx.Client(timeout=TIMEOUT) as client: response = client.post(self.API_URL, headers=headers, json=payload) response.raise_for_status() data = response.json() content_blocks = data.get("content", []) return "".join( block.get("text", "") for block in content_blocks if block.get("type") == "text" ) except httpx.HTTPStatusError as e: raise RuntimeError( f"Claude API error {e.response.status_code}: {e.response.text}" ) from e except httpx.RequestError as e: raise RuntimeError(f"Claude API request failed: {e}") from e class OpenAICompatibleProvider(LLMProvider): """OpenAI Chat Completions-compatible provider. Used for OpenAI, OpenRouter, and any custom OpenAI-compatible endpoint. Set base_url to point to any compatible API. """ DEFAULT_BASE_URL = "https://api.openai.com/v1" def __init__( self, api_key: str, model: str | None = None, base_url: str | None = None, default_model: str = "gpt-4o-mini", ): super().__init__(api_key, model or default_model) self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") def generate(self, prompt: str, system: str = "") -> str: url = f"{self.base_url}/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } messages: list[dict] = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) payload = { "model": self.model, "messages": messages, "max_tokens": 2048, } try: with httpx.Client(timeout=TIMEOUT) as client: response = client.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"] except httpx.HTTPStatusError as e: raise RuntimeError( f"API error {e.response.status_code}: {e.response.text}" ) from e except httpx.RequestError as e: raise RuntimeError(f"API request failed: {e}") from e class GeminiProvider(LLMProvider): """Google Gemini provider via Generative Language API.""" API_BASE = "https://generativelanguage.googleapis.com/v1beta/models" def __init__(self, api_key: str, model: str | None = None): super().__init__(api_key, model or DEFAULT_MODELS["gemini"]) def generate(self, prompt: str, system: str = "") -> str: url = f"{self.API_BASE}/{self.model}:generateContent" params = {"key": self.api_key} headers = {"Content-Type": "application/json"} parts: list[dict] = [] if system: parts.append({"text": f"{system}\n\n{prompt}"}) else: parts.append({"text": prompt}) payload = { "contents": [{"parts": parts}], "generationConfig": {"maxOutputTokens": 2048}, } try: with httpx.Client(timeout=TIMEOUT) as client: response = client.post(url, params=params, headers=headers, json=payload) response.raise_for_status() data = response.json() candidates = data.get("candidates", []) if not candidates: return "" content = candidates[0].get("content", {}) parts_out = content.get("parts", []) return "".join(part.get("text", "") for part in parts_out) except httpx.HTTPStatusError as e: raise RuntimeError( f"Gemini API error {e.response.status_code}: {e.response.text}" ) from e except httpx.RequestError as e: raise RuntimeError(f"Gemini API request failed: {e}") from e def get_llm_provider( provider_name: str, api_key: str, model: str | None = None, base_url: str | None = None, ) -> LLMProvider: """Factory: returns an LLMProvider instance. Args: provider_name: 'claude', 'openai', 'gemini', 'openrouter', or 'custom'. api_key: API key for the provider. model: Optional model override. base_url: Optional base URL override (used for 'openrouter' and 'custom'). Returns: An LLMProvider instance ready to generate text. """ name = provider_name.lower() if name == "claude": return ClaudeProvider(api_key=api_key, model=model) if name == "openai": return OpenAICompatibleProvider( api_key=api_key, model=model, base_url="https://api.openai.com/v1", default_model=DEFAULT_MODELS["openai"], ) if name == "gemini": return GeminiProvider(api_key=api_key, model=model) if name == "openrouter": return OpenAICompatibleProvider( api_key=api_key, model=model, base_url=base_url or "https://openrouter.ai/api/v1", default_model=DEFAULT_MODELS["openrouter"], ) if name == "custom": if not base_url: raise ValueError("Provider 'custom' richiede un base_url configurato nelle impostazioni.") return OpenAICompatibleProvider( api_key=api_key, model=model, base_url=base_url, default_model=model or "", ) raise ValueError( f"Provider LLM '{provider_name}' non supportato. " f"Usa: claude, openai, gemini, openrouter, custom." )