import json import re import traceback from typing import Any, Final, Protocol, cast import httpx import litellm from litellm._logging import _ENABLE_SECRET_REDACTION, _redact_string, verbose_logger from litellm.litellm_core_utils.secret_redaction import redact_string from litellm.types.utils import LlmProviders from ..exceptions import ( APIConnectionError, APIError, AuthenticationError, BadGatewayError, BadRequestError, ContentPolicyViolationError, ContextWindowExceededError, InternalServerError, NotFoundError, PermissionDeniedError, RateLimitError, ServiceUnavailableError, Timeout, UnprocessableEntityError, ) class ExceptionCheckers: """ Helper class for checking various error conditions in exception strings. """ @staticmethod def is_error_str_rate_limit(error_str: str, status_code: int | None = None) -> bool: """ Check if an error string indicates a rate limit error. Args: error_str: The error string to check status_code: The HTTP status the provider returned, when known. Gates only the bare-number branch: providers echo the request back in validation errors and 429 is an ordinary token id, so an echoed prompt can put a standalone 429 in the body of a 400. The phrase branches stay ungated (#11455). Returns: True if the error indicates a rate limit, False otherwise """ if not isinstance(error_str, str): return False # A standalone 429 counts unless the provider's own status says otherwise. The # status is read off an arbitrary exception, so a non-integer means "unknown". if re.search(r"\b429\b", error_str) and (not isinstance(status_code, int) or status_code == 429): return True _error_str_lower: Final = error_str.lower() # Match "rate limit" (including variations like rate-limit / rate_limit) if re.search(r"rate[\s_\-]*limit", _error_str_lower): return True ####################################### # Mistral API returns this error string ######################################### if "service tier capacity exceeded" in _error_str_lower: return True return False @staticmethod def is_error_str_context_window_exceeded(error_str: str) -> bool: """ Check if an error string indicates a context window exceeded error. """ _error_str_lowercase: Final = error_str.lower() # Exclude param validation errors (e.g. OpenAI "user" param max 64 chars) if "string_above_max_length" in _error_str_lowercase: return False if "invalid 'user'" in _error_str_lowercase and "string too long" in _error_str_lowercase: return False known_exception_substrings: Final = [ "exceed context limit", "this model's maximum context length is", "string too long. expected a string with maximum length", "model's maximum context limit", "is longer than the model's context length", "input tokens exceed the configured limit", "`inputs` tokens + `max_new_tokens` must be", "exceeds the available context size", # llama.cpp/Lemonade "exceeds the maximum number of tokens allowed", # Gemini ] for substring in known_exception_substrings: if substring in _error_str_lowercase: return True # Cerebras pattern: "Current length is X while limit is Y" if "current length is" in _error_str_lowercase and "while limit is" in _error_str_lowercase: return True if "maximum input length is" in _error_str_lowercase and "tokens" in _error_str_lowercase: return True return False @staticmethod def is_azure_content_policy_violation_error(error_str: str) -> bool: """ Check if an error string indicates a content policy violation error. """ _lower: Final = error_str.lower() known_exception_substrings: Final = [ "content_policy_violation", "responsibleaipolicyviolation", "the response was filtered due to the prompt triggering azure openai's content management", "your task failed as a result of our safety system", "the model produced invalid content", "content_filter_policy", "your request was rejected as a result of our safety system", ] for substring in known_exception_substrings: if substring in _lower: return True return False def get_error_message(error_obj) -> str | None: """ OpenAI Returns Error message that is nested, this extract the message Example: { 'request': "", 'message': "Error code: 400 - {\'error\': {\'message\': \"Invalid 'temperature': decimal above maximum value. Expected a value <= 2, but got 200 instead.\", 'type': 'invalid_request_error', 'param': 'temperature', 'code': 'decimal_above_max_value'}}", 'body': { 'message': "Invalid 'temperature': decimal above maximum value. Expected a value <= 2, but got 200 instead.", 'type': 'invalid_request_error', 'param': 'temperature', 'code': 'decimal_above_max_value' }, 'code': 'decimal_above_max_value', 'param': 'temperature', 'type': 'invalid_request_error', 'response': "", 'status_code': 400, 'request_id': 'req_f287898caa6364cd42bc01355f74dd2a' } """ try: # First, try to access the message directly from the 'body' key if error_obj is None: return None if hasattr(error_obj, "body"): _error_obj_body: Final = getattr(error_obj, "body") if isinstance(_error_obj_body, dict): # OpenAI-style: {"message": "...", "type": "...", ...} if _error_obj_body.get("message"): return _error_obj_body.get("message") # Azure-style: {"error": {"message": "...", ...}} nested_error: Final = _error_obj_body.get("error") if isinstance(nested_error, dict): return nested_error.get("message") # If all else fails, return None return None except Exception: return None ####### EXCEPTION MAPPING ################ def _get_body_error_code(error_str: str) -> int | None: """Return error.code from a JSON error body, or None if not parseable.""" try: body: Final = json.loads(error_str) code: Final = body.get("error", {}).get("code") return int(code) if code is not None else None except Exception: return None def _get_response_headers(original_exception: Exception) -> httpx.Headers | None: """ Extract and return the response headers from an exception, if present. Used for accurate retry logic. """ _response_headers: httpx.Headers | None = None try: _response_headers = getattr(original_exception, "headers", None) error_response: Final = getattr(original_exception, "response", None) if not _response_headers and error_response: _response_headers = getattr(error_response, "headers", None) if not _response_headers: _response_headers = getattr(original_exception, "litellm_response_headers", None) except Exception: return None return _response_headers def extract_and_raise_litellm_exception( response: Any | None, error_str: str, model: str, custom_llm_provider: str, ): """ Covers scenario where litellm sdk calling proxy. Enables raising the special errors raised by litellm, eg. ContextWindowExceededError. Relevant Issue: https://github.com/BerriAI/litellm/issues/7259 """ pattern: Final = r"litellm\.\w+Error" # Search for the exception in the error string match: Final = re.search(pattern, error_str) # Extract the exception if found if match: exception_name = match.group(0) exception_name = exception_name.strip().replace("litellm.", "") raised_exception_obj: Final = getattr(litellm, exception_name, None) if raised_exception_obj: # Try with response parameter first, fall back to without it # Some exceptions (e.g., APIConnectionError) don't accept response param try: raise raised_exception_obj( message=error_str, llm_provider=custom_llm_provider, model=model, response=response, ) except TypeError: # Exception doesn't accept response parameter raise raised_exception_obj( message=error_str, llm_provider=custom_llm_provider, model=model, ) class _ProviderHTTPException(Protocol): status_code: int message: str response: httpx.Response request: httpx.Request body: object code: str llm_provider: str def _map_openai_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: # custom_llm_provider is openai, make it OpenAI message = get_error_message(error_obj=original_exception) if message is None: if hasattr(original_exception, "message"): message = original_exception.message else: message = str(original_exception) if message is not None and isinstance( message, str ): # done to prevent user-confusion. Relevant issue - https://github.com/BerriAI/litellm/issues/1414 message = message.replace("OPENAI", custom_llm_provider.upper()) message = message.replace( "openai.OpenAIError", f"{custom_llm_provider}.{custom_llm_provider}Error", ) if custom_llm_provider == "openai": exception_provider = "OpenAI" + "Exception" else: exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception" if ExceptionCheckers.is_error_str_rate_limit( error_str, status_code=getattr(original_exception, "status_code", None) ): raise RateLimitError( message=f"RateLimitError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), ) elif ExceptionCheckers.is_error_str_context_window_exceeded(error_str): raise ContextWindowExceededError( message=f"ContextWindowExceededError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif "invalid_request_error" in error_str and "model_not_found" in error_str: raise NotFoundError( message=f"{exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif "A timeout occurred" in error_str: raise Timeout( message=f"{exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif ( ("invalid_request_error" in error_str and "content_policy_violation" in error_str) or ("Invalid prompt" in error_str and "violating our usage policy" in error_str) or ("request was rejected as a result of the safety system" in error_str.lower()) ): raise ContentPolicyViolationError( message=f"ContentPolicyViolationError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif "invalid_encrypted_content" in error_str or "could not be verified" in error_str: helpful_message: Final = ( f"{exception_provider} - {message}\n\n" " This error occurs when load balancing Responses API across deployments with different API keys.\n" " Encrypted content is tied to the organization that created it and cannot be decrypted by other organizations.\n\n" " Solution: Enable 'encrypted_content_affinity' to route follow-up requests to the correct deployment:\n\n" " router_settings:\n" " enable_pre_call_checks: true\n" " optional_pre_call_checks:\n" " - encrypted_content_affinity\n\n" " Learn more: https://docs.litellm.ai/docs/response_api#encrypted-content-affinity-multi-region-load-balancing" ) raise BadRequestError( message=helpful_message, llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, body=getattr(original_exception, "body", None), ) elif "invalid_request_error" in error_str and "Incorrect API key provided" not in error_str: raise BadRequestError( message=f"{exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, body=getattr(original_exception, "body", None), ) elif ( "Web server is returning an unknown error" in error_str or "The server had an error processing your request." in error_str ): raise litellm.InternalServerError( message=f"{exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, ) elif "Request too large" in error_str: raise RateLimitError( message=f"RateLimitError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif ( "The api_key client option must be set either by passing api_key to the client or by setting the OPENAI_API_KEY environment variable" in error_str ): raise AuthenticationError( message=f"AuthenticationError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif "Mistral API raised a streaming error" in error_str: _request: Final = httpx.Request(method="POST", url="https://api.openai.com/v1") raise APIError( status_code=500, message=f"{exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, request=_request, litellm_debug_info=extra_information, ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 400: raise BadRequestError( message=f"{exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"AuthenticationError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 404: raise NotFoundError( message=f"NotFoundError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 408: raise Timeout( message=f"Timeout Error: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 422: raise BadRequestError( message=f"{exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, body=getattr(original_exception, "body", None), ) elif original_exception.status_code == 429: raise RateLimitError( message=f"RateLimitError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 500: raise InternalServerError( message=f"InternalServerError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 502: raise BadGatewayError( message=f"BadGatewayError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"ServiceUnavailableError: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"Timeout Error: {exception_provider} - {message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=original_exception.status_code, ) else: raise APIError( status_code=original_exception.status_code, message=f"APIError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, request=getattr(original_exception, "request", None), litellm_debug_info=extra_information, ) else: # if no status code then it is an APIConnectionError: https://github.com/openai/openai-python#handling-errors # exception_mapping_worked = True raise APIConnectionError( message=f"APIConnectionError: {exception_provider} - {message}", llm_provider=custom_llm_provider, model=model, litellm_debug_info=extra_information, request=httpx.Request(method="POST", url="https://api.openai.com/v1/"), ) def _map_anthropic_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if ( "prompt is too long" in error_str or "prompt: length" in error_str or ExceptionCheckers.is_error_str_context_window_exceeded(error_str) ): raise ContextWindowExceededError( message=f"AnthropicError - {error_str}", model=model, llm_provider="anthropic", ) elif "overloaded_error" in error_str or "Overloaded" in error_str: raise InternalServerError( message=f"AnthropicError - {error_str}", model=model, llm_provider="anthropic", ) if "Invalid API Key" in error_str: raise AuthenticationError( message=f"AnthropicError - {error_str}", model=model, llm_provider="anthropic", ) if "content filtering policy" in error_str: raise ContentPolicyViolationError( message=f"AnthropicError - {error_str}", model=model, llm_provider="anthropic", ) if "Client error '400 Bad Request'" in error_str: raise BadRequestError( message=f"AnthropicError - {error_str}", model=model, llm_provider="anthropic", ) if hasattr(original_exception, "status_code"): verbose_logger.debug("status_code: %s", original_exception.status_code) if original_exception.status_code == 401: raise AuthenticationError( message=f"AnthropicException - {error_str}", llm_provider="anthropic", model=model, ) elif original_exception.status_code == 403: raise PermissionDeniedError( message=f"AnthropicException - {error_str}", llm_provider="anthropic", model=model, response=original_exception.response, ) elif original_exception.status_code == 400 or original_exception.status_code == 413: raise BadRequestError( message=f"AnthropicException - {error_str}", model=model, llm_provider="anthropic", ) elif original_exception.status_code == 404: raise NotFoundError( message=f"AnthropicException - {error_str}", model=model, llm_provider="anthropic", ) elif original_exception.status_code == 408: raise Timeout( message=f"AnthropicException - {error_str}", model=model, llm_provider="anthropic", ) elif original_exception.status_code == 429: raise RateLimitError( message=f"AnthropicException - {error_str}", llm_provider="anthropic", model=model, ) elif original_exception.status_code == 500 or original_exception.status_code == 529: raise litellm.InternalServerError( message=f"AnthropicException - {error_str}. Handle with `litellm.InternalServerError`.", llm_provider="anthropic", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 502: raise BadGatewayError( message=f"AnthropicException BadGatewayError - {error_str}", llm_provider="anthropic", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 503: raise litellm.ServiceUnavailableError( message=f"AnthropicException - {error_str}. Handle with `litellm.ServiceUnavailableError`.", llm_provider="anthropic", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"AnthropicException Timeout - {error_str}", model=model, llm_provider="anthropic", exception_status_code=original_exception.status_code, ) def _map_replicate_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "Incorrect authentication token" in error_str: raise AuthenticationError( message=f"ReplicateException - {error_str}", llm_provider="replicate", model=model, response=getattr(original_exception, "response", None), ) elif "input is too long" in error_str: raise ContextWindowExceededError( message=f"ReplicateException - {error_str}", model=model, llm_provider="replicate", response=getattr(original_exception, "response", None), ) elif exception_type == "ModelError": raise BadRequestError( message=f"ReplicateException - {error_str}", model=model, llm_provider="replicate", response=getattr(original_exception, "response", None), ) elif "Request was throttled" in error_str: raise RateLimitError( message=f"ReplicateException - {error_str}", llm_provider="replicate", model=model, response=getattr(original_exception, "response", None), ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 401: raise AuthenticationError( message=f"ReplicateException - {original_exception.message}", llm_provider="replicate", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 400 or original_exception.status_code == 413: raise BadRequestError( message=f"ReplicateException - {original_exception.message}", model=model, llm_provider="replicate", response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 422: raise UnprocessableEntityError( message=f"ReplicateException - {original_exception.message}", model=model, llm_provider="replicate", response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"ReplicateException - {original_exception.message}", model=model, llm_provider="replicate", ) elif original_exception.status_code == 429: raise RateLimitError( message=f"ReplicateException - {original_exception.message}", llm_provider="replicate", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 500: raise ServiceUnavailableError( message=f"ReplicateException - {original_exception.message}", llm_provider="replicate", model=model, response=getattr(original_exception, "response", None), ) raise APIError( status_code=500, message=f"ReplicateException - {original_exception}", llm_provider="replicate", model=model, request=httpx.Request( method="POST", url="https://api.replicate.com/v1/deployments", ), ) def _map_openai_like_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "authorization denied for" in error_str: # Predibase returns the raw API Key in the response - this block ensures it's not returned in the exception if error_str is not None and isinstance(error_str, str) and "bearer" in error_str.lower(): # only keep the first 10 chars after the occurnence of "bearer" _bearer_token_start_index: Final = error_str.lower().find("bearer") error_str = error_str[: _bearer_token_start_index + 14] error_str += "XXXXXXX" + '"' raise AuthenticationError( message=f"{custom_llm_provider.capitalize()}Exception: Authentication Error - {error_str}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif ExceptionCheckers.is_error_str_context_window_exceeded(error_str): raise ContextWindowExceededError( message=f"{custom_llm_provider.capitalize()}Exception: Context Window Error - {error_str}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif "token_quota_reached" in error_str: raise RateLimitError( message=f"{custom_llm_provider.capitalize()}Exception: Rate Limit Errror - {error_str}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), ) elif "The server received an invalid response from an upstream server." in error_str: raise litellm.InternalServerError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, ) elif "model_no_support_for_function" in error_str: raise BadRequestError( message=f"{custom_llm_provider.capitalize()}Exception - Use 'watsonx_text' route instead. IBM WatsonX does not support `/text/chat` endpoint. - {error_str}", llm_provider=custom_llm_provider, model=model, ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 500: raise litellm.InternalServerError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, ) elif original_exception.status_code == 403: raise PermissionDeniedError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, response=_response_or_stub(original_exception, status_code=403), ) elif original_exception.status_code == 400: raise BadRequestError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, ) elif original_exception.status_code == 404: raise NotFoundError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", llm_provider=custom_llm_provider, model=model, ) elif original_exception.status_code == 408: raise Timeout( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 422 or original_exception.status_code == 424: raise BadRequestError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 429: raise RateLimitError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=original_exception.status_code, ) _BEDROCK_MANTLE_CONTEXT_WINDOW_PATTERN: Final = re.compile(r"prompt tokens \((\d+)\) exceed model maximum \((\d+)\)") def _get_bedrock_mantle_context_window_message(error_str: str) -> str | None: """ Mantle reports context overflow as a structured validation error rather than the plain-text patterns Bedrock itself uses, so it needs its own detection and a message clients recognize as context overflow (litellm/litellm#36546). """ if "invalid_request_error" not in error_str and "validation_error" not in error_str: return None match = _BEDROCK_MANTLE_CONTEXT_WINDOW_PATTERN.search(error_str) if match is None: return None prompt_tokens, max_tokens = match.groups() return f"prompt is too long: {prompt_tokens} tokens > {max_tokens} maximum" def _map_bedrock_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if custom_llm_provider == "bedrock_mantle": mantle_context_window_message = _get_bedrock_mantle_context_window_message(error_str) if mantle_context_window_message is not None: raise ContextWindowExceededError( message=mantle_context_window_message, model=model, llm_provider=custom_llm_provider, ) if ( "too many tokens" in error_str or "expected maxLength:" in error_str or "Input is too long" in error_str or "prompt is too long" in error_str or "prompt: length: 1.." in error_str or "Too many input tokens" in error_str ): raise ContextWindowExceededError( message=f"BedrockException: Context Window Error - {error_str}", model=model, llm_provider="bedrock", ) elif "Conversation blocks and tool result blocks cannot be provided in the same turn." in error_str: raise BadRequestError( message=f"BedrockException - {error_str}\n. Enable 'litellm.modify_params=True' (for PROXY do: `litellm_settings::modify_params: True`) to insert a dummy assistant message and fix this error.", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif "Malformed input request" in error_str: raise BadRequestError( message=f"BedrockException - {error_str}", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif "A conversation must start with a user message." in error_str: raise BadRequestError( message=f"BedrockException - {error_str}\n. Pass in default user message via `completion(..,user_continue_message=)` or enable `litellm.modify_params=True`.\nFor Proxy: do via `litellm_settings::modify_params: True` or user_continue_message under `litellm_params`", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif ( "Unable to locate credentials" in error_str or "The security token included in the request is invalid" in error_str ): raise AuthenticationError( message=f"BedrockException Invalid Authentication - {error_str}", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif "AccessDeniedException" in error_str: raise PermissionDeniedError( message=f"BedrockException PermissionDeniedError - {error_str}", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif "throttlingException" in error_str or "ThrottlingException" in error_str: raise RateLimitError( message=f"BedrockException: Rate Limit Error - {error_str}", model=model, llm_provider="bedrock", response=getattr(original_exception, "response", None), ) elif "Connect timeout on endpoint URL" in error_str or "timed out" in error_str: raise Timeout( message=f"BedrockException: Timeout Error - {error_str}", model=model, llm_provider="bedrock", ) elif "Could not process image" in error_str: raise litellm.InternalServerError( message=f"BedrockException - {error_str}", model=model, llm_provider="bedrock", ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 500: raise ServiceUnavailableError( message=f"BedrockException - {original_exception.message}", llm_provider="bedrock", model=model, response=httpx.Response( status_code=500, request=httpx.Request(method="POST", url="https://api.openai.com/v1/"), ), ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"BedrockException - {original_exception.message}", llm_provider="bedrock", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 400: raise BadRequestError( message=f"BedrockException - {original_exception.message}", llm_provider="bedrock", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 404: raise NotFoundError( message=f"BedrockException - {original_exception.message}", llm_provider="bedrock", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"BedrockException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 422: raise BadRequestError( message=f"BedrockException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 429: raise RateLimitError( message=f"BedrockException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"BedrockException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"BedrockException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=original_exception.status_code, ) def _map_sagemaker_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "Unable to locate credentials" in error_str: raise BadRequestError( message=f"litellm.BadRequestError: SagemakerException - {error_str}", model=model, llm_provider="sagemaker", response=getattr(original_exception, "response", None), ) elif "Input validation error: `best_of` must be > 0 and <= 2" in error_str: raise BadRequestError( message="SagemakerException - the value of 'n' must be > 0 and <= 2 for sagemaker endpoints", model=model, llm_provider="sagemaker", response=getattr(original_exception, "response", None), ) elif ( "`inputs` tokens + `max_new_tokens` must be <=" in error_str or "instance type with more CPU capacity or memory" in error_str ): raise ContextWindowExceededError( message=f"SagemakerException - {error_str}", model=model, llm_provider="sagemaker", response=getattr(original_exception, "response", None), ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 500: raise ServiceUnavailableError( message=f"SagemakerException - {original_exception.message}", llm_provider=custom_llm_provider, model=model, response=httpx.Response( status_code=500, request=httpx.Request(method="POST", url="https://api.openai.com/v1/"), ), ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"SagemakerException - {original_exception.message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 400: raise BadRequestError( message=f"SagemakerException - {original_exception.message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 404: raise NotFoundError( message=f"SagemakerException - {original_exception.message}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"SagemakerException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 422 or original_exception.status_code == 424: raise BadRequestError( message=f"SagemakerException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 429: raise RateLimitError( message=f"SagemakerException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"SagemakerException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"SagemakerException - {original_exception.message}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=original_exception.status_code, ) def _map_vertex_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "Vertex AI API has not been used in project" in error_str or "Unable to find your project" in error_str: raise BadRequestError( message=f"litellm.BadRequestError: {custom_llm_provider}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, response=httpx.Response( status_code=400, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), litellm_debug_info=extra_information, ) if "400 Request payload size exceeds" in error_str: raise ContextWindowExceededError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, ) elif ExceptionCheckers.is_error_str_context_window_exceeded(error_str): raise ContextWindowExceededError( message=f"ContextWindowExceededError: {custom_llm_provider.capitalize()}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif "None Unknown Error." in error_str or "Content has no parts." in error_str: raise litellm.InternalServerError( message=f"litellm.InternalServerError: {custom_llm_provider}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, response=httpx.Response( status_code=500, content=str(original_exception), request=httpx.Request(method="completion", url="https://github.com/BerriAI/litellm"), ), litellm_debug_info=extra_information, ) elif "API key not valid." in error_str: raise AuthenticationError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif "403" in error_str: raise BadRequestError( message=f"{custom_llm_provider.capitalize()}Exception BadRequestError - {error_str}", model=model, llm_provider=custom_llm_provider, response=httpx.Response( status_code=403, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), litellm_debug_info=extra_information, ) elif ( "The response was blocked." in error_str or "Output blocked by content filtering policy" in error_str # anthropic on vertex ai ): raise ContentPolicyViolationError( message=f"{custom_llm_provider.capitalize()}Exception ContentPolicyViolationError - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=400, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), ) elif ( "429 Quota exceeded" in error_str or "Quota exceeded for" in error_str or "Resource exhausted" in error_str or "IndexError: list index out of range" in error_str or "429 Unable to submit request because the service is temporarily out of capacity." in error_str ): raise RateLimitError( message=f"litellm.RateLimitError: {custom_llm_provider}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=429, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), ) elif ( isinstance(getattr(original_exception, "status_code", None), int) and 500 <= original_exception.status_code < 600 and _get_body_error_code(error_str) == 429 ): # upstream gateway wraps a 429 inside a 5xx envelope # e.g. HTTP 500/503 with {"error":{"code":429,...}}. # Scoped to 5xx so HTTP 400/401 with body code:429 # still maps to BadRequestError / AuthenticationError. raise RateLimitError( message=f"litellm.RateLimitError: {custom_llm_provider}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=429, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), ) elif "500 Internal Server Error" in error_str or "The model is overloaded." in error_str: raise litellm.InternalServerError( message=f"litellm.InternalServerError: {custom_llm_provider}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) if hasattr(original_exception, "status_code"): if original_exception.status_code == 400: raise BadRequestError( message=f"{custom_llm_provider.capitalize()}Exception BadRequestError - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=400, request=httpx.Request( method="POST", url="https://cloud.google.com/vertex-ai/", ), ), ) if original_exception.status_code == 401: raise AuthenticationError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, ) if original_exception.status_code == 403: raise PermissionDeniedError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, response=httpx.Response( status_code=403, request=httpx.Request( method="POST", url="https://cloud.google.com/vertex-ai/", ), ), ) if original_exception.status_code == 404: raise NotFoundError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, ) if original_exception.status_code == 408: raise Timeout( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, ) if original_exception.status_code == 429: raise RateLimitError( message=f"litellm.RateLimitError: {custom_llm_provider.capitalize()}Exception - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=429, request=httpx.Request( method="POST", url=" https://cloud.google.com/vertex-ai/", ), ), ) if original_exception.status_code == 500: raise litellm.InternalServerError( message=f"{custom_llm_provider.capitalize()}Exception InternalServerError - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, response=httpx.Response( status_code=500, content=str(original_exception), request=httpx.Request(method="completion", url="https://github.com/BerriAI/litellm"), ), ) if original_exception.status_code == 502: raise APIConnectionError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, ) if original_exception.status_code == 503: raise ServiceUnavailableError( message=f"{custom_llm_provider.capitalize()}Exception - {error_str}", llm_provider=custom_llm_provider, model=model, ) def _map_cloudflare_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "Authentication error" in error_str: raise AuthenticationError( message=f"Cloudflare Exception - {original_exception.message}", llm_provider="cloudflare", model=model, response=getattr(original_exception, "response", None), ) if "must have required property" in error_str: raise BadRequestError( message=f"Cloudflare Exception - {original_exception.message}", llm_provider="cloudflare", model=model, response=getattr(original_exception, "response", None), ) def _map_cohere_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "invalid api token" in error_str or "No API key provided." in error_str: raise AuthenticationError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif "invalid type: parameter" in error_str: raise BadRequestError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif "too many tokens" in error_str: raise ContextWindowExceededError( message=f"CohereException - {original_exception.message}", model=model, llm_provider="cohere", response=getattr(original_exception, "response", None), ) elif "internal server error" in error_str.lower(): raise InternalServerError( message=f"CohereException - {error_str}", model=model, llm_provider="cohere", response=getattr(original_exception, "response", None), ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 400 or original_exception.status_code == 498: raise BadRequestError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, ) elif original_exception.status_code == 500: raise InternalServerError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif ( "CohereConnectionError" in exception_type ): # cohere seems to fire these errors when we load test it (1k+ messages / min) raise RateLimitError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif "invalid type:" in error_str: raise BadRequestError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) elif "Unexpected server error" in error_str: raise InternalServerError( message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, response=getattr(original_exception, "response", None), ) else: if hasattr(original_exception, "status_code"): raise APIError( status_code=original_exception.status_code, message=f"CohereException - {original_exception.message}", llm_provider="cohere", model=model, request=getattr(original_exception, "request", None), ) raise cast(Exception, original_exception) def _map_huggingface_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "length limit exceeded" in error_str: raise ContextWindowExceededError( message=error_str, model=model, llm_provider="huggingface", response=getattr(original_exception, "response", None), ) elif "A valid user token is required" in error_str: raise BadRequestError( message=error_str, llm_provider="huggingface", model=model, response=getattr(original_exception, "response", None), ) elif "Rate limit reached" in error_str: raise RateLimitError( message=error_str, llm_provider="huggingface", model=model, response=getattr(original_exception, "response", None), ) if hasattr(original_exception, "status_code"): if original_exception.status_code == 401: raise AuthenticationError( message=f"HuggingfaceException - {original_exception.message}", llm_provider="huggingface", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 400: raise BadRequestError( message=f"HuggingfaceException - {original_exception.message}", model=model, llm_provider="huggingface", response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"HuggingfaceException - {original_exception.message}", model=model, llm_provider="huggingface", ) elif original_exception.status_code == 429: raise RateLimitError( message=f"HuggingfaceException - {original_exception.message}", llm_provider="huggingface", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"HuggingfaceException - {original_exception.message}", llm_provider="huggingface", model=model, response=getattr(original_exception, "response", None), ) else: raise APIError( status_code=original_exception.status_code, message=f"HuggingfaceException - {original_exception.message}", llm_provider="huggingface", model=model, request=getattr(original_exception, "request", None), ) def _map_ai21_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if hasattr(original_exception, "message"): if "Prompt has too many tokens" in original_exception.message: raise ContextWindowExceededError( message=f"AI21Exception - {original_exception.message}", model=model, llm_provider="ai21", response=getattr(original_exception, "response", None), ) if "Bad or missing API token." in original_exception.message: raise BadRequestError( message=f"AI21Exception - {original_exception.message}", model=model, llm_provider="ai21", response=getattr(original_exception, "response", None), ) if hasattr(original_exception, "status_code"): if original_exception.status_code == 401: raise AuthenticationError( message=f"AI21Exception - {original_exception.message}", llm_provider="ai21", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"AI21Exception - {original_exception.message}", model=model, llm_provider="ai21", ) if original_exception.status_code == 422: raise BadRequestError( message=f"AI21Exception - {original_exception.message}", model=model, llm_provider="ai21", response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 429: raise RateLimitError( message=f"AI21Exception - {original_exception.message}", llm_provider="ai21", model=model, response=getattr(original_exception, "response", None), ) else: raise APIError( status_code=original_exception.status_code, message=f"AI21Exception - {original_exception.message}", llm_provider="ai21", model=model, request=getattr(original_exception, "request", None), ) def _map_nlp_cloud_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "detail" in error_str: if "Input text length should not exceed" in error_str: raise ContextWindowExceededError( message=f"NLPCloudException - {error_str}", model=model, llm_provider="nlp_cloud", response=getattr(original_exception, "response", None), ) elif "value is not a valid" in error_str: raise BadRequestError( message=f"NLPCloudException - {error_str}", model=model, llm_provider="nlp_cloud", response=getattr(original_exception, "response", None), ) else: raise APIError( status_code=500, message=f"NLPCloudException - {error_str}", model=model, llm_provider="nlp_cloud", request=getattr(original_exception, "request", None), ) if hasattr(original_exception, "status_code"): # https://docs.nlpcloud.com/?shell#errors if ( original_exception.status_code == 400 or original_exception.status_code == 406 or original_exception.status_code == 413 or original_exception.status_code == 422 ): raise BadRequestError( message=f"NLPCloudException - {original_exception.message}", llm_provider="nlp_cloud", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 401 or original_exception.status_code == 403: raise AuthenticationError( message=f"NLPCloudException - {original_exception.message}", llm_provider="nlp_cloud", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 522 or original_exception.status_code == 524: raise Timeout( message=f"NLPCloudException - {original_exception.message}", model=model, llm_provider="nlp_cloud", ) elif original_exception.status_code == 429 or original_exception.status_code == 402: raise RateLimitError( message=f"NLPCloudException - {original_exception.message}", llm_provider="nlp_cloud", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 500 or original_exception.status_code == 503: raise APIError( status_code=original_exception.status_code, message=f"NLPCloudException - {original_exception.message}", llm_provider="nlp_cloud", model=model, request=getattr(original_exception, "request", None), ) elif original_exception.status_code == 504 or original_exception.status_code == 520: raise ServiceUnavailableError( message=f"NLPCloudException - {original_exception.message}", model=model, llm_provider="nlp_cloud", response=getattr(original_exception, "response", None), ) else: raise APIError( status_code=original_exception.status_code, message=f"NLPCloudException - {original_exception.message}", llm_provider="nlp_cloud", model=model, request=getattr(original_exception, "request", None), ) def _map_together_ai_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: try: error_response = json.loads(error_str) except Exception: error_response = {"error": error_str} if "error" in error_response and "`inputs` tokens + `max_new_tokens` must be <=" in error_response["error"]: raise ContextWindowExceededError( message=f"TogetherAIException - {error_response['error']}", model=model, llm_provider="together_ai", response=getattr(original_exception, "response", None), ) elif "error" in error_response and "invalid private key" in error_response["error"]: raise AuthenticationError( message=f"TogetherAIException - {error_response['error']}", llm_provider="together_ai", model=model, response=getattr(original_exception, "response", None), ) elif "error" in error_response and "INVALID_ARGUMENT" in error_response["error"]: raise BadRequestError( message=f"TogetherAIException - {error_response['error']}", model=model, llm_provider="together_ai", response=getattr(original_exception, "response", None), ) elif "A timeout occurred" in error_str: raise Timeout( message=f"TogetherAIException - {error_str}", model=model, llm_provider="together_ai", ) elif ( "error" in error_response and "API key doesn't match expected format." in error_response["error"] or "error_type" in error_response and error_response["error_type"] == "validation" ): raise BadRequestError( message=f"TogetherAIException - {error_response['error']}", model=model, llm_provider="together_ai", response=getattr(original_exception, "response", None), ) if hasattr(original_exception, "status_code"): if original_exception.status_code == 408: raise Timeout( message=f"TogetherAIException - {original_exception.message}", model=model, llm_provider="together_ai", ) elif original_exception.status_code == 422: raise BadRequestError( message=f"TogetherAIException - {error_response['error']}", model=model, llm_provider="together_ai", response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 429: raise RateLimitError( message=f"TogetherAIException - {original_exception.message}", llm_provider="together_ai", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 524: raise Timeout( message=f"TogetherAIException - {original_exception.message}", llm_provider="together_ai", model=model, ) else: raise APIError( status_code=original_exception.status_code, message=f"TogetherAIException - {original_exception.message}", llm_provider="together_ai", model=model, request=getattr(original_exception, "request", None), ) def _map_aleph_alpha_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if "This is longer than the model's maximum context length" in error_str: raise ContextWindowExceededError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, response=getattr(original_exception, "response", None), ) elif "InvalidToken" in error_str or "No token provided" in error_str: raise BadRequestError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, response=getattr(original_exception, "response", None), ) elif hasattr(original_exception, "status_code"): verbose_logger.debug("status code: %s", original_exception.status_code) if original_exception.status_code == 401: raise AuthenticationError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, ) elif original_exception.status_code == 400: raise BadRequestError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 429: raise RateLimitError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 500: raise ServiceUnavailableError( message=f"AlephAlphaException - {original_exception.message}", llm_provider="aleph_alpha", model=model, response=getattr(original_exception, "response", None), ) raise cast(Exception, original_exception) raise cast(Exception, original_exception) def _map_ollama_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if isinstance(original_exception, dict): error_str = original_exception.get("error", "") else: error_str = str(original_exception) if "no such file or directory" in error_str: raise BadRequestError( message=f"OllamaException: Invalid Model/Model not loaded - {original_exception}", model=model, llm_provider="ollama", response=getattr(original_exception, "response", None), ) elif "Failed to establish a new connection" in error_str: raise ServiceUnavailableError( message=f"OllamaException: {original_exception}", llm_provider="ollama", model=model, response=getattr(original_exception, "response", None), ) elif "Invalid response object from API" in error_str: raise BadRequestError( message=f"OllamaException: {original_exception}", llm_provider="ollama", model=model, response=getattr(original_exception, "response", None), ) elif "Read timed out" in error_str: raise Timeout( message=f"OllamaException: {original_exception}", llm_provider="ollama", model=model, ) def _map_vllm_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if hasattr(original_exception, "status_code"): if original_exception.status_code == 0: raise APIConnectionError( message=f"VLLMException - {original_exception.message}", llm_provider="vllm", model=model, request=getattr(original_exception, "request", None), ) def _map_azure_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: message = get_error_message(error_obj=original_exception) if message is None: if hasattr(original_exception, "message"): message = original_exception.message else: message = str(original_exception) # Azure OpenAI (especially Images) often nests error details under # body["error"]. Detect content policy violations using the structured # payload in addition to string matching. azure_error_code: str | None = None try: body_dict: Final = getattr(original_exception, "body", None) or {} if isinstance(body_dict, dict): if isinstance(body_dict.get("error"), dict): azure_error_code = body_dict["error"].get("code") # Also check inner_error for # ResponsibleAIPolicyViolation which indicates a # content policy violation even when the top-level # code is generic (e.g. "invalid_request_error"). if azure_error_code != "content_policy_violation": _inner: Final = body_dict["error"].get("inner_error") or body_dict["error"].get("innererror") if isinstance(_inner, dict) and _inner.get("code") == "ResponsibleAIPolicyViolation": azure_error_code = "content_policy_violation" else: azure_error_code = body_dict.get("code") except Exception: azure_error_code = None if "Internal server error" in error_str: raise litellm.InternalServerError( message=f"AzureException Internal server error - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif "This model's maximum context length is" in error_str: raise ContextWindowExceededError( message=f"AzureException ContextWindowExceededError - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif "DeploymentNotFound" in error_str: raise NotFoundError( message=f"AzureException NotFoundError - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif azure_error_code == "content_policy_violation" or ExceptionCheckers.is_azure_content_policy_violation_error( error_str ): from litellm.llms.azure.exception_mapping import ( AzureOpenAIExceptionMapping, ) raise AzureOpenAIExceptionMapping.create_content_policy_violation_error( message=message, model=model, extra_information=extra_information, original_exception=original_exception, ) elif azure_error_code == "invalid_encrypted_content" or "could not be verified" in error_str: helpful_message: Final = ( f"AzureException - {message}\n\n" "This error occurs when load balancing Responses API across deployments with different API keys.\n" " Encrypted content is tied to the organization that created it and cannot be decrypted by other organizations.\n\n" " Solution: Enable 'encrypted_content_affinity' to route follow-up requests to the correct deployment:\n\n" " router_settings:\n" " enable_pre_call_checks: true\n" " optional_pre_call_checks:\n" " - encrypted_content_affinity\n\n" " Learn more: https://docs.litellm.ai/docs/response_api#encrypted-content-affinity-multi-region-load-balancing" ) raise BadRequestError( message=helpful_message, llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), body=getattr(original_exception, "body", None), ) elif "invalid_request_error" in error_str and getattr(original_exception, "status_code", None) in (None, 400): raise BadRequestError( message=f"AzureException BadRequestError - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), body=getattr(original_exception, "body", None), ) elif "The api_key client option must be set either by passing api_key to the client or by setting" in error_str: raise AuthenticationError( message=f"{exception_provider} AuthenticationError - {message}", llm_provider=custom_llm_provider, model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif "Connection error" in error_str: raise APIConnectionError( message=f"{exception_provider} APIConnectionError - {message}", llm_provider=custom_llm_provider, model=model, litellm_debug_info=extra_information, ) elif hasattr(original_exception, "status_code"): if original_exception.status_code == 400: raise BadRequestError( message=f"AzureException - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), body=getattr(original_exception, "body", None), ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"AzureException AuthenticationError - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 404: raise NotFoundError( message=f"AzureException NotFoundError - {message}", llm_provider="azure", model=model, litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 408: raise Timeout( message=f"AzureException Timeout - {message}", model=model, litellm_debug_info=extra_information, llm_provider="azure", ) elif original_exception.status_code == 422: raise BadRequestError( message=f"AzureException BadRequestError - {message}", model=model, llm_provider="azure", litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 429: raise RateLimitError( message=f"AzureException RateLimitError - {message}", model=model, llm_provider="azure", litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 502: raise BadGatewayError( message=f"AzureException BadGatewayError - {message}", model=model, llm_provider="azure", litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"AzureException ServiceUnavailableError - {message}", model=model, llm_provider="azure", litellm_debug_info=extra_information, response=getattr(original_exception, "response", None), ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"AzureException Timeout - {message}", model=model, litellm_debug_info=extra_information, llm_provider="azure", exception_status_code=original_exception.status_code, ) else: raise APIError( status_code=original_exception.status_code, message=f"AzureException APIError - {message}", llm_provider="azure", litellm_debug_info=extra_information, model=model, request=httpx.Request(method="POST", url="https://openai.com/"), ) else: # if no status code then it is an APIConnectionError: https://github.com/openai/openai-python#handling-errors raise APIConnectionError( message=f"{exception_provider} APIConnectionError - {message}\n{_redact_string(traceback.format_exc())}", llm_provider="azure", model=model, litellm_debug_info=extra_information, request=httpx.Request(method="POST", url="https://openai.com/"), ) def _map_openrouter_exception( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_type: str, exception_provider: str, extra_information: str, ) -> None: if hasattr(original_exception, "status_code"): if original_exception.status_code == 400: raise BadRequestError( message=f"{exception_provider} - {error_str}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 401: raise AuthenticationError( message=f"AuthenticationError: {exception_provider} - {error_str}", llm_provider=custom_llm_provider, model=model, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 404: raise NotFoundError( message=f"NotFoundError: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 408: raise Timeout( message=f"Timeout Error: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) elif original_exception.status_code == 422: raise BadRequestError( message=f"BadRequestError: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 429: raise RateLimitError( message=f"RateLimitError: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 503: raise ServiceUnavailableError( message=f"ServiceUnavailableError: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), litellm_debug_info=extra_information, ) elif original_exception.status_code == 504: # gateway timeout error raise Timeout( message=f"Timeout Error: {exception_provider} - {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=original_exception.status_code, ) else: raise APIError( status_code=original_exception.status_code, message=f"APIError: {exception_provider} - {error_str}", llm_provider=custom_llm_provider, model=model, request=getattr(original_exception, "request", None), litellm_debug_info=extra_information, ) else: # if no status code then it is an APIConnectionError: https://github.com/openai/openai-python#handling-errors raise APIConnectionError( message=f"APIConnectionError: {exception_provider} - {error_str}", llm_provider=custom_llm_provider, model=model, litellm_debug_info=extra_information, request=httpx.Request(method="POST", url="https://api.openai.com/v1/"), ) def _response_or_stub(original_exception: _ProviderHTTPException, status_code: int) -> httpx.Response: response: Final = original_exception.response if hasattr(original_exception, "response") else None if response is not None: return response return httpx.Response( status_code=status_code, request=httpx.Request(method="POST", url="https://docs.litellm.ai/docs") ) def _map_exception_by_status( *, model: str, original_exception: _ProviderHTTPException, custom_llm_provider: str, error_str: str, exception_provider: str, extra_information: str, ) -> None: status_code: Final = original_exception.status_code if hasattr(original_exception, "status_code") else None if not isinstance(status_code, int) or status_code < 400: return if getattr(original_exception, "status_code_is_synthesized", False): return message: Final = f"{exception_provider} - {error_str}" response: Final = original_exception.response if hasattr(original_exception, "response") else None match status_code: case 401: raise AuthenticationError( message=message, llm_provider=custom_llm_provider, model=model, response=response, litellm_debug_info=extra_information, ) case 403: raise PermissionDeniedError( message=message, llm_provider=custom_llm_provider, model=model, response=_response_or_stub(original_exception, status_code=status_code), litellm_debug_info=extra_information, ) case 404: raise NotFoundError( message=message, model=model, llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, ) case 408: raise Timeout( message=message, model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) case 429: raise RateLimitError( message=message, model=model, llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, ) case 500: raise InternalServerError( message=message, llm_provider=custom_llm_provider, model=model, response=response, litellm_debug_info=extra_information, ) case 502: raise BadGatewayError( message=message, llm_provider=custom_llm_provider, model=model, response=response, litellm_debug_info=extra_information, ) case 503: raise ServiceUnavailableError( message=message, llm_provider=custom_llm_provider, model=model, response=response, litellm_debug_info=extra_information, ) case 504: raise Timeout( message=message, model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, exception_status_code=status_code, ) case _ if status_code < 500: raise BadRequestError( message=message, model=model, llm_provider=custom_llm_provider, response=response, litellm_debug_info=extra_information, ) case _: raise APIError( status_code=status_code, message=message, llm_provider=custom_llm_provider, model=model, request=original_exception.request if hasattr(original_exception, "request") else None, litellm_debug_info=extra_information, ) def exception_type( model, original_exception, custom_llm_provider, completion_kwargs={}, extra_kwargs={}, ): """Maps an LLM Provider Exception to OpenAI Exception Format""" if any(isinstance(original_exception, exc_type) for exc_type in litellm.LITELLM_EXCEPTION_TYPES): return original_exception exception_mapping_worked = False exception_provider = custom_llm_provider mappable_exception: Final[_ProviderHTTPException] = cast("_ProviderHTTPException", original_exception) if litellm.suppress_debug_info is False: print() # noqa: T201 print( # noqa: T201 "\033[1;31mGive Feedback / Get Help: https://github.com/BerriAI/litellm/issues/new\033[0m" ) print( # noqa: T201 "LiteLLM.Info: If you need to debug this error, use `litellm._turn_on_debug()'." ) print() # noqa: T201 litellm_response_headers: Final = _get_response_headers(original_exception=original_exception) try: error_str = redact_string(str(original_exception)) if _ENABLE_SECRET_REDACTION else str(original_exception) extra_information = "" if model or custom_llm_provider: if hasattr(original_exception, "message"): error_str = ( redact_string(str(original_exception.message)) if _ENABLE_SECRET_REDACTION else str(original_exception.message) ) if isinstance(original_exception, BaseException): exception_type = type(original_exception).__name__ else: exception_type = "" ################################################################################ # Common Extra information needed for all providers # We pass num retries, api_base, vertex_deployment etc to the exception here ################################################################################ try: _api_base: Final = litellm.get_api_base(model=model, optional_params=extra_kwargs) messages: Final = litellm.get_first_chars_messages(kwargs=completion_kwargs) _vertex_project: Final = extra_kwargs.get("vertex_project") _vertex_location: Final = extra_kwargs.get("vertex_location") _metadata: Final = extra_kwargs.get("metadata", {}) or {} _model_group: Final = _metadata.get("model_group") _deployment: Final = _metadata.get("deployment") extra_information = f"\nModel: {model}" if isinstance(custom_llm_provider, str) and len(custom_llm_provider) > 0: exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception" if _api_base: extra_information += f"\nAPI Base: `{_api_base}`" if messages and len(messages) > 0 and litellm.redact_messages_in_exceptions is False: extra_information += f"\nMessages: `{messages}`" if _model_group is not None: extra_information += f"\nmodel_group: `{_model_group}`\n" if _deployment is not None: extra_information += f"\ndeployment: `{_deployment}`\n" if _vertex_project is not None: extra_information += f"\nvertex_project: `{_vertex_project}`\n" if _vertex_location is not None: extra_information += f"\nvertex_location: `{_vertex_location}`\n" # on litellm proxy add key name + team to exceptions extra_information = _add_key_name_and_team_to_alert(request_info=extra_information, metadata=_metadata) except Exception: # DO NOT LET this Block raising the original exception pass ################################################################################ # End of Common Extra information Needed for all providers ################################################################################ ################################################################################ #################### Start of Provider Exception mapping #################### ################################################################################ if ( "Request Timeout Error" in error_str or "Request timed out" in error_str or "Timed out generating response" in error_str or "The read operation timed out" in error_str ): exception_mapping_worked = True raise Timeout( message=f"APITimeoutError - Request timed out. Error_str: {error_str}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, ) if ( custom_llm_provider == "litellm_proxy" ): # handle special case where calling litellm proxy + exception str contains error message extract_and_raise_litellm_exception( response=getattr(original_exception, "response", None), error_str=error_str, model=model, custom_llm_provider=custom_llm_provider, ) if ( custom_llm_provider == "openai" or custom_llm_provider == "text-completion-openai" or custom_llm_provider == "custom_openai" or custom_llm_provider in litellm.openai_compatible_providers or custom_llm_provider == "mistral" or custom_llm_provider == "runwayml" ): _map_openai_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "anthropic" or custom_llm_provider == "anthropic_text": # one of the anthropics _map_anthropic_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "replicate": _map_replicate_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider in litellm._openai_like_providers: _map_openai_like_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider in ("bedrock", "bedrock_mantle"): _map_bedrock_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "sagemaker" or custom_llm_provider == "sagemaker_chat": _map_sagemaker_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif ( custom_llm_provider == LlmProviders.VERTEX_AI or custom_llm_provider == LlmProviders.VERTEX_AI_BETA or custom_llm_provider == LlmProviders.GEMINI ): _map_vertex_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "cloudflare": _map_cloudflare_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "cohere" or custom_llm_provider == "cohere_chat": # Cohere _map_cohere_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "huggingface": _map_huggingface_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "ai21": _map_ai21_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "nlp_cloud": _map_nlp_cloud_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "together_ai": _map_together_ai_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "aleph_alpha": _map_aleph_alpha_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "ollama" or custom_llm_provider == "ollama_chat": _map_ollama_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "vllm": _map_vllm_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) elif custom_llm_provider == "azure" or custom_llm_provider == "azure_text": _map_azure_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) if custom_llm_provider == "openrouter": _map_openrouter_exception( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_type=exception_type, exception_provider=exception_provider, extra_information=extra_information, ) if "BadRequestError.__init__() missing 1 required positional argument: 'param'" in str( original_exception ): # deal with edge-case invalid request error bug in openai-python sdk exception_mapping_worked = True raise BadRequestError( message=f"{exception_provider} BadRequestError : This can happen due to missing AZURE_API_VERSION: {original_exception}", model=model, llm_provider=custom_llm_provider, response=getattr(original_exception, "response", None), ) else: # ensure generic errors always return APIConnectionError= """ For unmapped exceptions - raise the exception with traceback - https://github.com/BerriAI/litellm/issues/4201 """ exception_mapping_worked = True _map_exception_by_status( model=model, original_exception=mappable_exception, custom_llm_provider=custom_llm_provider, error_str=error_str, exception_provider=exception_provider, extra_information=extra_information, ) if hasattr(original_exception, "request"): raise APIConnectionError( message=f"{exception_provider} - {error_str}", llm_provider=custom_llm_provider, model=model, request=getattr(original_exception, "request", None), ) else: raise APIConnectionError( message=f"{original_exception}\n{_redact_string(traceback.format_exc())}", llm_provider=custom_llm_provider, model=model, request=httpx.Request(method="POST", url="https://api.openai.com/v1/"), # stub the request ) except Exception as e: # LOGGING exception_logging( logger_fn=None, additional_args={ "exception_mapping_worked": exception_mapping_worked, "original_exception": original_exception, }, exception=e, ) # don't let an error with mapping interrupt the user from receiving an error from the llm api calls if exception_mapping_worked: setattr(e, "litellm_response_headers", litellm_response_headers) raise e else: for error_type in litellm.LITELLM_EXCEPTION_TYPES: if isinstance(e, error_type): setattr(e, "litellm_response_headers", litellm_response_headers) raise e # it's already mapped raised_exc: Final = APIConnectionError( message=f"{original_exception}\n{_redact_string(traceback.format_exc())}", llm_provider="", model="", ) setattr(raised_exc, "litellm_response_headers", litellm_response_headers) raise raised_exc ####### LOGGING ################### def exception_logging( additional_args={}, logger_fn=None, exception=None, ): try: model_call_details: Final = {} if exception: model_call_details["exception"] = exception model_call_details["additional_args"] = additional_args # User Logging -> if you pass in a custom logging function or want to use sentry breadcrumbs verbose_logger.debug( "Logging Details: logger_fn - %s | callable(logger_fn) - %s", logger_fn, callable(logger_fn) ) if logger_fn and callable(logger_fn): try: logger_fn( model_call_details ) # Expectation: any logger function passed in by the user should accept a dict object except Exception: verbose_logger.debug( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", traceback.format_exc() ) except Exception: verbose_logger.debug( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", traceback.format_exc() ) def _add_key_name_and_team_to_alert(request_info: str, metadata: dict) -> str: """ Internal helper function for litellm proxy Add the Key Name + Team Name to the error Only gets added if the metadata contains the user_api_key_alias and user_api_key_team_alias [Non-Blocking helper function] """ try: _api_key_name: Final = metadata.get("user_api_key_alias", None) _user_api_key_team_alias: Final = metadata.get("user_api_key_team_alias", None) if _api_key_name is not None: request_info = f"\n\nKey Name: `{_api_key_name}`\nTeam: `{_user_api_key_team_alias}`" + request_info return request_info except Exception: return request_info