e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
161 lines
4.0 KiB
Python
161 lines
4.0 KiB
Python
"""
|
|
LLM Service Exceptions
|
|
======================
|
|
|
|
Custom exception classes for the LLM service.
|
|
Provides a consistent exception hierarchy for better error handling.
|
|
Maintains parity with upstream dev branch.
|
|
"""
|
|
|
|
|
|
class LLMError(Exception):
|
|
"""Base exception for all LLM-related errors."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
details: dict[str, object] | None = None,
|
|
provider: str | None = None,
|
|
):
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.details = details or {}
|
|
self.provider = provider
|
|
|
|
def __str__(self) -> str:
|
|
provider_prefix = f"[{self.provider}] " if self.provider else ""
|
|
if self.details:
|
|
return f"{provider_prefix}{self.message} (details: {self.details})"
|
|
return f"{provider_prefix}{self.message}"
|
|
|
|
|
|
class LLMConfigError(LLMError):
|
|
"""Raised when there's an error in LLM configuration."""
|
|
|
|
pass
|
|
|
|
|
|
class LLMProviderError(LLMError):
|
|
"""Raised when there's an error with the LLM provider."""
|
|
|
|
pass
|
|
|
|
|
|
class LLMCircuitBreakerError(LLMError):
|
|
"""Raised when circuit breaker blocks LLM execution."""
|
|
|
|
pass
|
|
|
|
|
|
class LLMAPIError(LLMError):
|
|
"""
|
|
Raised when an API call to an LLM provider fails.
|
|
Standardizes status_code and provider name.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
status_code: int | None = None,
|
|
provider: str | None = None,
|
|
details: dict[str, object] | None = None,
|
|
):
|
|
super().__init__(message, details, provider)
|
|
self.status_code = status_code
|
|
|
|
def __str__(self) -> str:
|
|
parts = []
|
|
if self.provider:
|
|
parts.append(f"[{self.provider}]")
|
|
if self.status_code:
|
|
parts.append(f"HTTP {self.status_code}")
|
|
parts.append(self.message)
|
|
return " ".join(parts)
|
|
|
|
|
|
class LLMTimeoutError(LLMAPIError):
|
|
"""Raised when an API call times out."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Request timed out",
|
|
timeout: float | None = None,
|
|
provider: str | None = None,
|
|
):
|
|
super().__init__(message, status_code=408, provider=provider)
|
|
self.timeout = timeout
|
|
|
|
|
|
class LLMRateLimitError(LLMAPIError):
|
|
"""Raised when rate limited by the API."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Rate limit exceeded",
|
|
retry_after: float | None = None,
|
|
provider: str | None = None,
|
|
):
|
|
super().__init__(message, status_code=429, provider=provider)
|
|
self.retry_after = retry_after
|
|
|
|
|
|
class LLMAuthenticationError(LLMAPIError):
|
|
"""Raised when authentication fails (invalid API key, etc.)."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Authentication failed",
|
|
provider: str | None = None,
|
|
):
|
|
super().__init__(message, status_code=401, provider=provider)
|
|
|
|
|
|
class LLMModelNotFoundError(LLMAPIError):
|
|
"""Raised when the requested model is not found."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Model not found",
|
|
model: str | None = None,
|
|
provider: str | None = None,
|
|
):
|
|
super().__init__(message, status_code=404, provider=provider)
|
|
self.model = model
|
|
|
|
|
|
class LLMParseError(LLMError):
|
|
"""Raised when parsing LLM output fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Failed to parse LLM output",
|
|
provider: str | None = None,
|
|
details: dict[str, object] | None = None,
|
|
):
|
|
super().__init__(message, details=details, provider=provider)
|
|
|
|
|
|
# Multi-provider specific aliases for mapping rules
|
|
class ProviderQuotaExceededError(LLMRateLimitError):
|
|
pass
|
|
|
|
|
|
class ProviderContextWindowError(LLMAPIError):
|
|
pass
|
|
|
|
|
|
__all__ = [
|
|
"LLMError",
|
|
"LLMConfigError",
|
|
"LLMProviderError",
|
|
"LLMCircuitBreakerError",
|
|
"LLMAPIError",
|
|
"LLMTimeoutError",
|
|
"LLMRateLimitError",
|
|
"LLMAuthenticationError",
|
|
"LLMModelNotFoundError",
|
|
"LLMParseError",
|
|
"ProviderQuotaExceededError",
|
|
"ProviderContextWindowError",
|
|
]
|