Spaces:
Sleeping
Sleeping
File size: 3,418 Bytes
74711df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | """Application-level error taxonomy for the LLM layer.
This module normalizes provider exceptions into a small stable error contract
that service-layer code can rely on without depending on provider-specific
classes.
"""
from functools import lru_cache
class LLMError(Exception):
"""Base exception for all LLM-layer failures."""
def __init__(self, message: str | None = None):
self.message = message or self.__doc__
super().__init__(self.message.strip())
class LLMConfigError(LLMError):
"""Model configuration is invalid or missing."""
class LLMProviderError(LLMError):
"""Raised for provider-side failures that are not more specific."""
class LLMTimeoutError(LLMProviderError):
"""Raised when an LLM request times out."""
class LLMRateLimitError(LLMProviderError):
"""Raised when the provider rejects requests due to rate limits."""
class LLMStructuredOutputError(LLMError):
"""The model's response could not be parsed to JSON."""
class LLMSchemaValidationError(LLMError):
"""Parsed JSON schema failed validation."""
class LLMRetryExhaustedError(LLMError):
"""All retries and fallbacks were exhausted."""
@lru_cache(maxsize=1)
def _litellm_exception_types() -> tuple[type[Exception], ...]:
"""Returns LiteLLM exception base types exposed by the SDK."""
try:
import litellm
except ImportError:
return ()
known = getattr(litellm, "LITELLM_EXCEPTION_TYPES", ())
return tuple(exc for exc in known if isinstance(exc, type) and issubclass(exc, Exception))
def coerce_provider_exception(error: BaseException) -> BaseException:
"""Normalizes provider exceptions into app-level error types.
Args:
error: Exception instance raised by provider/runtime code.
Returns:
BaseException: Mapped application exception when recognized, otherwise
the original exception.
"""
if not isinstance(error, Exception):
return error
try:
import litellm
except ImportError:
return error
timeout_type = getattr(litellm, "Timeout", None)
rate_limit_type = getattr(litellm, "RateLimitError", None)
if isinstance(timeout_type, type) and isinstance(error, timeout_type):
return LLMTimeoutError(str(error))
if isinstance(rate_limit_type, type) and isinstance(error, rate_limit_type):
return LLMRateLimitError(str(error))
if _litellm_exception_types() and isinstance(error, _litellm_exception_types()):
return LLMProviderError(str(error))
return error
def error_type_for_exception(error: BaseException) -> str:
"""Maps exceptions to normalized metadata error labels.
Args:
error: Exception to classify.
Returns:
str: Error-category label suitable for telemetry metadata.
"""
normalized = coerce_provider_exception(error)
if isinstance(normalized, LLMTimeoutError):
return "timeout"
if isinstance(normalized, LLMRateLimitError):
return "rate_limit"
if isinstance(normalized, LLMProviderError):
return "provider"
if isinstance(normalized, LLMStructuredOutputError):
return "structured_output"
if isinstance(normalized, LLMSchemaValidationError):
return "schema_validation"
if isinstance(normalized, LLMRetryExhaustedError):
return "retry_exhausted"
return normalized.__class__.__name__.lower()
|