From f70b5208ecbdad79c35cb9caa05cd8538f5bf6b8 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sat, 6 Jun 2026 17:10:30 +0545 Subject: [PATCH] refactor(devgen/providers): shared provider base class Abstract out generate validation and error handling into BaseProvider, eliminating duplicated logic across Anthropic, OpenRouter, and OpenAI providers. --- devgen/providers/anthropic.py | 38 +++++--------- devgen/providers/base.py | 90 +++++++++++++++++++++++++++++++++ devgen/providers/gemini.py | 89 ++++++++++++++------------------ devgen/providers/huggingface.py | 73 +++++++++----------------- devgen/providers/ollama.py | 63 +++++++++++------------ devgen/providers/openai.py | 74 +++++---------------------- devgen/providers/openrouter.py | 55 +++++++------------- 7 files changed, 223 insertions(+), 259 deletions(-) create mode 100644 devgen/providers/base.py diff --git a/devgen/providers/anthropic.py b/devgen/providers/anthropic.py index 1516398..0ee17fb 100644 --- a/devgen/providers/anthropic.py +++ b/devgen/providers/anthropic.py @@ -1,33 +1,19 @@ import anthropic -from devgen.utils import format_token_limit_error, is_token_limit_error +from devgen.providers.base import BaseProvider -class AnthropicProvider: +class AnthropicProvider(BaseProvider): """Generates content using Anthropic's Claude models.""" - def generate( - self, prompt: str, api_key: str, model: str = "claude-3-opus-20240229", **kwargs - ) -> str: - """Generates a response using the Anthropic API.""" - if not api_key: - raise ValueError( - "Anthropic API key is missing. " - "Set it via `devgen setup config` or pass --api-key." - ) + DISPLAY_NAME = "Anthropic" + DEFAULT_MODEL = "claude-3-opus-20240229" - try: - client = anthropic.Anthropic(api_key=api_key) - message = client.messages.create( - model=model, - max_tokens=1024, - messages=[{"role": "user", "content": prompt}], - ) - return message.content[0].text.strip() - except Exception as e: - if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("Anthropic", e)) from e - raise RuntimeError( - f"Anthropic generation failed: {e}. " - "Check the model id and your account credits." - ) from e + def _generate(self, prompt, api_key, model, **kwargs): + client = anthropic.Anthropic(api_key=api_key) + message = client.messages.create( + model=model, + max_tokens=1024, + messages=[{"role": "user", "content": prompt}], + ) + return message.content[0].text diff --git a/devgen/providers/base.py b/devgen/providers/base.py new file mode 100644 index 0000000..61aa2e1 --- /dev/null +++ b/devgen/providers/base.py @@ -0,0 +1,90 @@ +"""Base class for AI providers. + +Every provider implements :meth:`_generate` and inherits the standard +contract: API-key validation, token-limit detection, and friendly +RuntimeError wrapping. Providers with provider-specific error shapes +(eg. Hugging Face's HTTPError / RequestException split) override +:meth:`_handle_error`. +""" + +from abc import ABC, abstractmethod +from typing import Any + +from devgen.utils import format_token_limit_error, is_token_limit_error + + +class BaseProvider(ABC): + """Abstract base for all AI providers.""" + + #: Human-readable name used in error messages. + DISPLAY_NAME: str = "" + #: Whether the provider requires an API key. + REQUIRES_API_KEY: bool = True + #: Default model id when the user doesn't specify one. + DEFAULT_MODEL: str = "" + + def generate( + self, + prompt: str, + api_key: str | None = None, + model: str | None = None, + **kwargs: Any, + ) -> str: + """Public entry point. Validates input, dispatches, and wraps errors. + + Args: + prompt: The prompt to send to the model. + api_key: Provider API key. May be ``None`` for local providers + (see :attr:`REQUIRES_API_KEY`). + model: Model id. Falls back to :attr:`DEFAULT_MODEL` when ``None``. + **kwargs: Forwarded to :meth:`_generate`. + + Returns: + The generated text, stripped. + + Raises: + ValueError: If the provider requires an API key and none was given. + RuntimeError: For any underlying failure, with a friendly message. + """ + if self.REQUIRES_API_KEY and not api_key: + raise ValueError( + f"{self.DISPLAY_NAME} API key is missing. " + "Set it via `devgen setup config` or pass --api-key." + ) + chosen_model = model or self.DEFAULT_MODEL + try: + return self._generate(prompt, api_key, chosen_model, **kwargs).strip() + except Exception as e: + self._handle_error(e) + + def _handle_error(self, error: Exception) -> None: + """Translate an internal exception into a user-facing RuntimeError. + + Token-limit / context-window errors are reformatted with + :func:`format_token_limit_error`; everything else is wrapped in a + generic provider message. Subclasses override this to add + provider-specific error shapes (eg. HTTPError vs RequestException). + """ + if is_token_limit_error(error): + raise RuntimeError( + format_token_limit_error(self.DISPLAY_NAME, error) + ) from error + raise RuntimeError( + f"{self.DISPLAY_NAME} request failed: {error}. " + "Check the model name, API key permissions, and your account quota." + ) from error + + @abstractmethod + def _generate( + self, + prompt: str, + api_key: str | None, + model: str, + **kwargs: Any, + ) -> str: + """Provider-specific generation. Return the raw generated text. + + Implementations may raise any exception; :meth:`_handle_error` will + translate it. Re-raise an exception to defer to default handling. + """ + raise NotImplementedError diff --git a/devgen/providers/gemini.py b/devgen/providers/gemini.py index 4b25feb..3779c84 100644 --- a/devgen/providers/gemini.py +++ b/devgen/providers/gemini.py @@ -1,31 +1,22 @@ import time + from google import genai from google.genai import types -from devgen.utils import format_token_limit_error, is_token_limit_error +from devgen.providers.base import BaseProvider +from devgen.utils import is_token_limit_error -class GeminiProvider: +class GeminiProvider(BaseProvider): """Generates content using Google's Gemini models.""" - def generate( - self, prompt: str, api_key: str, model: str = "gemini-2.5-flash", **kwargs - ) -> str: - """Generates a response using the Gemini API. - - Default model is updated to gemini-2.5-flash which is the latest - recommended model. - """ - if not api_key: - raise ValueError( - "Gemini API key is missing. " - "Set it via `devgen setup config` or pass --api-key." - ) - - client = genai.Client(api_key=api_key) + DISPLAY_NAME = "Gemini" + DEFAULT_MODEL = "gemini-2.5-flash" + def _generate(self, prompt, api_key, model, **kwargs): max_retries = kwargs.get("max_retries", 5) - base_delay = kwargs.get("retry_delay", 10) # 10 seconds base delay for 429 + base_delay = kwargs.get("retry_delay", 10) # 10s base for 429 backoff + client = genai.Client(api_key=api_key) for attempt in range(max_retries + 1): try: @@ -39,38 +30,34 @@ class GeminiProvider: max_output_tokens=kwargs.get("max_output_tokens", 2048), ), ) - return response.text.strip() + return response.text except Exception as e: - error_msg = str(e).upper() - # Token / context-window overflow is not retryable — no point - # backing off, the prompt is just too big. if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("Gemini", e)) from e - # Check for 429 or ResourceExhausted in common error formats - if any( - x in error_msg - for x in [ - "429", - "RESOURCE_EXHAUSTED", - "QUOTA_EXCEEDED", - "THROTTLED", - ] - ): - if attempt < max_retries: - delay = base_delay * (2**attempt) - print( - f"\n[Gemini] Quota limit hit (429). This is often a temporary rate limit (RPM). " - f"Waiting {delay}s and retrying... (Attempt {attempt + 1}/{max_retries})" - ) - time.sleep(delay) - continue - else: - raise RuntimeError( - "Gemini API quota exhausted (429) after multiple retries.\n\n" - "POSSIBLE CAUSES:\n" - "1. Rate Limit (RPM/RPD): You are sending requests too fast for the Free Tier.\n" - "2. Daily Limit: You have reached the maximum free requests allowed for today (typically 1,500).\n" - "3. Credit/Billing: If on a paid plan, your credit may be exhausted.\n\n" - "Please check your usage metrics at: https://aistudio.google.com/app/usage" - ) from e - raise RuntimeError(f"Gemini generation failed: {e}") from e + raise # base class handles with friendly message + if self._is_quota_error(e) and attempt < max_retries: + delay = base_delay * (2**attempt) + print( + f"\n[Gemini] Quota limit hit (429). This is often a temporary " + f"rate limit (RPM). Waiting {delay}s and retrying... " + f"(Attempt {attempt + 1}/{max_retries})" + ) + time.sleep(delay) + continue + if self._is_quota_error(e): + raise RuntimeError( + "Gemini API quota exhausted (429) after multiple retries.\n\n" + "POSSIBLE CAUSES:\n" + "1. Rate Limit (RPM/RPD): You are sending requests too fast for the Free Tier.\n" + "2. Daily Limit: You have reached the maximum free requests allowed for today (typically 1,500).\n" + "3. Credit/Billing: If on a paid plan, your credit may be exhausted.\n\n" + "Please check your usage metrics at: https://aistudio.google.com/app/usage" + ) from e + raise + + @staticmethod + def _is_quota_error(error: Exception) -> bool: + text = str(error).upper() + return any( + marker in text + for marker in ("429", "RESOURCE_EXHAUSTED", "QUOTA_EXCEEDED", "THROTTLED") + ) diff --git a/devgen/providers/huggingface.py b/devgen/providers/huggingface.py index 90dc4d2..7bac92f 100644 --- a/devgen/providers/huggingface.py +++ b/devgen/providers/huggingface.py @@ -1,34 +1,19 @@ import requests +from devgen.providers.base import BaseProvider from devgen.utils import format_token_limit_error, is_token_limit_error -class HuggingfaceProvider: +class HuggingfaceProvider(BaseProvider): """Generates content using Hugging Face Inference API.""" + DISPLAY_NAME = "Hugging Face" API_URL_TEMPLATE = "https://api-inference.huggingface.co/models/{model}" + DEFAULT_MODEL = "mistralai/Mistral-7B-Instruct-v0.2" - def generate( - self, - prompt: str, - api_key: str, - model: str = "mistralai/Mistral-7B-Instruct-v0.2", - **kwargs, - ) -> str: - """Generates a response using Hugging Face API.""" - if not api_key: - raise ValueError( - "Hugging Face API token is missing. " - "Set it via `devgen setup config` or pass --api-key." - ) - + def _generate(self, prompt, api_key, model, **kwargs): api_url = self.API_URL_TEMPLATE.format(model=model) headers = {"Authorization": f"Bearer {api_key}"} - - # HF models often expect specific prompting formats, but we'll send raw prompt - # Some models are text-generation, some are conversational. - # Assuming text-generation for generic usage. - payload = { "inputs": prompt, "parameters": {"max_new_tokens": 500, "return_full_text": False}, @@ -38,31 +23,12 @@ class HuggingfaceProvider: response = requests.post(api_url, headers=headers, json=payload) response.raise_for_status() result = response.json() - - if isinstance(result, list) and result and "generated_text" in result[0]: - return result[0]["generated_text"].strip() - elif isinstance(result, dict) and "error" in result: - err = result["error"] - if "is currently loading" in str(err).lower(): - raise RuntimeError( - f"Hugging Face model is loading: {err}. " - "Wait a minute and retry, or pick a different model." - ) - if is_token_limit_error(err): - raise RuntimeError( - format_token_limit_error("Hugging Face", err) - ) from err - raise RuntimeError(f"Hugging Face API error: {err}") - else: - raise RuntimeError( - "Hugging Face returned an unexpected response shape. " - "The model may not support text-generation." - ) - except requests.HTTPError as e: - status = e.response.status_code if e.response is not None else "?" if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("Hugging Face", e)) from e + raise RuntimeError( + format_token_limit_error(self.DISPLAY_NAME, e) + ) from e + status = e.response.status_code if e.response is not None else "?" raise RuntimeError( f"Hugging Face request failed (HTTP {status}): {e}. " "Check the model id and your token permissions." @@ -71,9 +37,18 @@ class HuggingfaceProvider: raise RuntimeError( f"Hugging Face network error: {e}. Check your connection." ) from e - except RuntimeError: - raise - except Exception as e: - if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("Hugging Face", e)) from e - raise RuntimeError(f"Hugging Face generation failed: {e}") from e + + if isinstance(result, list) and result and "generated_text" in result[0]: + return result[0]["generated_text"] + if isinstance(result, dict) and "error" in result: + err = result["error"] + if "is currently loading" in str(err).lower(): + raise RuntimeError( + f"Hugging Face model is loading: {err}. " + "Wait a minute and retry, or pick a different model." + ) + raise RuntimeError(f"Hugging Face API error: {err}") + raise RuntimeError( + "Hugging Face returned an unexpected response shape. " + "The model may not support text-generation." + ) diff --git a/devgen/providers/ollama.py b/devgen/providers/ollama.py index 577b07c..7c3afad 100644 --- a/devgen/providers/ollama.py +++ b/devgen/providers/ollama.py @@ -1,54 +1,51 @@ +from openai import OpenAI + +from devgen.providers.base import BaseProvider from devgen.utils import format_token_limit_error, is_token_limit_error -class OllamaProvider: +class OllamaProvider(BaseProvider): """Generates content using a local Ollama server. Ollama exposes an OpenAI-compatible endpoint at /v1/chat/completions, - so we can reuse the openai client. The host defaults to - http://localhost:11434 but can be overridden via the - `ollama_host` kwarg (or `ollama_host` in ~/.devgen.yaml). + so we reuse the ``openai`` client. The host defaults to + http://localhost:11434 but can be overridden via the ``ollama_host`` + kwarg (or ``ollama_host`` in ~/.devgen.yaml). """ + DISPLAY_NAME = "Ollama" + REQUIRES_API_KEY = False DEFAULT_HOST = "http://localhost:11434" DEFAULT_MODEL = "llama3.2" - def generate( - self, - prompt: str, - api_key: str | None = None, - model: str | None = None, - **kwargs, - ) -> str: - from openai import OpenAI - + def _generate(self, prompt, api_key, model, **kwargs): host = kwargs.get("ollama_host") or self.DEFAULT_HOST base_url = f"{host.rstrip('/')}/v1" - chosen_model = model or self.DEFAULT_MODEL - - client = OpenAI(base_url=base_url, api_key=api_key or "ollama") - kwargs.pop("debug", None) kwargs.pop("ollama_host", None) - try: - response = client.chat.completions.create( - model=chosen_model, - messages=[{"role": "user", "content": prompt}], - **kwargs, - ) - except Exception as e: - if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("Ollama", e)) from e - raise RuntimeError( - f"Ollama request failed (host={host}, model={chosen_model}): {e}. " - "Is the Ollama server running? Start it with `ollama serve` and " - "pull a model with `ollama pull `." - ) from e + client = OpenAI(base_url=base_url, api_key=api_key or "ollama") + response = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + **kwargs, + ) if not response.choices: raise RuntimeError( - f"Ollama returned no choices for model {chosen_model!r}. " + f"Ollama returned no choices for model {model!r}. " "The model may not be installed locally." ) - return response.choices[0].message.content.strip() + return response.choices[0].message.content + + def _handle_error(self, error): + """Override default wrapping — Ollama has no API key to check.""" + if is_token_limit_error(error): + raise RuntimeError( + format_token_limit_error(self.DISPLAY_NAME, error) + ) from error + raise RuntimeError( + f"{self.DISPLAY_NAME} request failed: {error}. " + "Is the Ollama server running? Start it with `ollama serve` and " + "pull a model with `ollama pull `." + ) from error diff --git a/devgen/providers/openai.py b/devgen/providers/openai.py index 4bf155d..466362e 100644 --- a/devgen/providers/openai.py +++ b/devgen/providers/openai.py @@ -1,70 +1,20 @@ from openai import OpenAI -from devgen.utils import format_token_limit_error, is_token_limit_error +from devgen.providers.base import BaseProvider -class OpenaiProvider: - """Generates a response string using the OpenAI ChatCompletion API based on the provided prompt and parameters. This method initializes an OpenAI client with the given API key, sends a chat completion request with specified model and additional parameters, and returns the content of the generated message. - - Args: - prompt (str): The input prompt to generate a response for. - api_key (str): The API key used to authenticate with the OpenAI service. - model (str, optional): The model to use for generation; defaults to "gpt-4o". - **kwargs: Additional keyword arguments to customize the API request (e.g., temperature). - - Returns: - str: The content of the generated response message. - """ +class OpenaiProvider(BaseProvider): + """Generates content using OpenAI's Chat Completions API.""" + DISPLAY_NAME = "OpenAI" DEFAULT_MODEL = "gpt-4o" - def generate( - self, prompt: str, api_key: str, model: str | None = None, **kwargs - ) -> str: - """Generates a response from the OpenAI ChatCompletion API based on the provided prompt and parameters. - - Creates a client instance with the specified API key, sends a chat completion request using the selected model and additional parameters, and returns the generated message content as a string. - - Args: - prompt (str): Prompt input. - api_key (str): OpenAI API key. - model (str, optional): Model to use (default: gpt-4o). - **kwargs: Additional OpenAI ChatCompletion parameters (e.g., temperature). - - Returns: - str: Generated response content. - """ - if not api_key: - raise ValueError( - "OpenAI API key is missing. " - "Set it via `devgen setup config` or pass --api-key." - ) - - # 1. Create a client instance with the API key. - try: - client = OpenAI(api_key=api_key) - except Exception as e: - # Add error handling if the client fails to initialize - raise RuntimeError( - f"Failed to initialize OpenAI client: {e}. " - "Verify the key is valid and that the `openai` package is installed." - ) from e - - # Remove debug from kwargs if present + def _generate(self, prompt, api_key, model, **kwargs): kwargs.pop("debug", None) - - # 2. Use the modern API syntax: client.chat.completions.create - try: - response = client.chat.completions.create( - model=model or self.DEFAULT_MODEL, - messages=[{"role": "user", "content": prompt}], - **kwargs, - ) - except Exception as e: - if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("OpenAI", e)) from e - raise RuntimeError( - f"OpenAI request failed: {e}. " - "Check the model name, API key permissions, and your account quota." - ) from e - return response.choices[0].message.content.strip() + client = OpenAI(api_key=api_key) + response = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + **kwargs, + ) + return response.choices[0].message.content diff --git a/devgen/providers/openrouter.py b/devgen/providers/openrouter.py index 3eea592..6c98296 100644 --- a/devgen/providers/openrouter.py +++ b/devgen/providers/openrouter.py @@ -1,46 +1,25 @@ from openai import OpenAI -from devgen.utils import format_token_limit_error, is_token_limit_error +from devgen.providers.base import BaseProvider -class OpenrouterProvider: +class OpenrouterProvider(BaseProvider): """Generates content using OpenRouter (OpenAI-compatible API).""" + DISPLAY_NAME = "OpenRouter" BASE_URL = "https://openrouter.ai/api/v1" + DEFAULT_MODEL = "openai/gpt-3.5-turbo" - def generate( - self, prompt: str, api_key: str, model: str = "openai/gpt-3.5-turbo", **kwargs - ) -> str: - """Generates a response using OpenRouter.""" - if not api_key: - raise ValueError( - "OpenRouter API key is missing. " - "Set it via `devgen setup config` or pass --api-key." - ) - - try: - client = OpenAI( - base_url=self.BASE_URL, - api_key=api_key, - ) - - # Remove debug from kwargs if present - kwargs.pop("debug", None) - - response = client.chat.completions.create( - model=model, - messages=[{"role": "user", "content": prompt}], - extra_headers={ - "HTTP-Referer": "https://github.com/S4NKALP/devgen", # Optional - "X-Title": "devgen CLI", # Optional - }, - **kwargs, - ) - return response.choices[0].message.content.strip() - except Exception as e: - if is_token_limit_error(e): - raise RuntimeError(format_token_limit_error("OpenRouter", e)) from e - raise RuntimeError( - f"OpenRouter request failed: {e}. " - "Verify the model id (e.g. `openai/gpt-4o`) and your API key credits." - ) from e + def _generate(self, prompt, api_key, model, **kwargs): + kwargs.pop("debug", None) + client = OpenAI(base_url=self.BASE_URL, api_key=api_key) + response = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + extra_headers={ + "HTTP-Referer": "https://github.com/S4NKALP/devgen", + "X-Title": "devgen CLI", + }, + **kwargs, + ) + return response.choices[0].message.content