Adds support for returning Azure Content Policy error information when exceptions from Azure OpenAI occur (#16231)

* add provider_specific_fields to ContentPolicyViolationError

* use provider_specific_fields in ProxyException

* update openai_exception_handler

* fix use exception checker for content policy violation azure

* add AzureOpenAIExceptionMapping

* test_azure_with_content_safety_error

* Accessing Provider-Specific Error Details

* TestExceptionCheckers

* unit test got provider_specific_fields=

* add clear types for error dict

* fix test_azure_with_content_safety_error

---------

Co-authored-by: Ishaan Jaff <ishaanjaffer0324@gmail.com>

docs/my-website/docs/exception_mapping.md

@@ -112,6 +112,85 @@ except openai.APITimeoutError as e:
     print(f"should_retry: {should_retry}")
 ```
 
+## Advanced
+
+### Accessing Provider-Specific Error Details
+
+LiteLLM exceptions include a `provider_specific_fields` attribute that contains additional error information specific to each provider. This is particularly useful for Azure OpenAI, which provides detailed content filtering information.
+
+#### Azure OpenAI - Content Policy Violation Inner Error Access
+
+When Azure OpenAI returns content policy violations, you can access the detailed content filtering results through the `innererror` field:
+
+```python
+import litellm
+from litellm.exceptions import ContentPolicyViolationError
+
+try:
+    response = litellm.completion(
+        model="azure/gpt-4",
+        messages=[
+            {
+                "role": "user", 
+                "content": "Some content that might violate policies"
+            }
+        ]
+    )
+except ContentPolicyViolationError as e:
+    # Access Azure-specific error details
+    if e.provider_specific_fields and "innererror" in e.provider_specific_fields:
+        innererror = e.provider_specific_fields["innererror"]
+        
+        # Access content filter results
+        content_filter_result = innererror.get("content_filter_result", {})
+        
+        print(f"Content filter code: {innererror.get('code')}")
+        print(f"Hate filtered: {content_filter_result.get('hate', {}).get('filtered')}")
+        print(f"Violence severity: {content_filter_result.get('violence', {}).get('severity')}")
+        print(f"Sexual content filtered: {content_filter_result.get('sexual', {}).get('filtered')}")
+```
+
+**Example Response Structure:**
+
+When calling the LiteLLM proxy, content policy violations will return detailed filtering information:
+
+```json
+{
+  "error": {
+    "message": "litellm.ContentPolicyViolationError: AzureException - The response was filtered due to the prompt triggering Azure OpenAI's content management policy...",
+    "type": null,
+    "param": null,
+    "code": "400",
+    "provider_specific_fields": {
+      "innererror": {
+        "code": "ResponsibleAIPolicyViolation",
+        "content_filter_result": {
+          "hate": {
+            "filtered": true,
+            "severity": "high"
+          },
+          "jailbreak": {
+            "filtered": false,
+            "detected": false
+          },
+          "self_harm": {
+            "filtered": false,
+            "severity": "safe"
+          },
+          "sexual": {
+            "filtered": false,
+            "severity": "safe"
+          },
+          "violence": {
+            "filtered": true,
+            "severity": "medium"
+          }
+        }
+      }
+    }
+  }
+}
+
 ## Details 
 
 To see how it's implemented - [check out the code](https://github.com/BerriAI/litellm/blob/a42c197e5a6de56ea576c73715e6c7c6b19fa249/litellm/utils.py#L1217)

litellm/exceptions.py

@@ -450,6 +450,7 @@ class ContentPolicyViolationError(BadRequestError):  # type: ignore
         llm_provider,
         response: Optional[httpx.Response] = None,
         litellm_debug_info: Optional[str] = None,
+        provider_specific_fields: Optional[dict] = None,
     ):
         self.status_code = 400
         self.message = "litellm.ContentPolicyViolationError: {}".format(message)
@@ -458,6 +459,8 @@ class ContentPolicyViolationError(BadRequestError):  # type: ignore
         self.litellm_debug_info = litellm_debug_info
         request = httpx.Request(method="POST", url="https://api.openai.com/v1")
         self.response = httpx.Response(status_code=400, request=request)
+        self.provider_specific_fields = provider_specific_fields
+        
         super().__init__(
             message=self.message,
             model=self.model,  # type: ignore
@@ -465,16 +468,18 @@ class ContentPolicyViolationError(BadRequestError):  # type: ignore
             response=self.response,
             litellm_debug_info=self.litellm_debug_info,
         )  # Call the base class constructor with the parameters it needs
+    
 
     def __str__(self):
-        _message = self.message
-        if self.num_retries:
-            _message += f" LiteLLM Retried: {self.num_retries} times"
-        if self.max_retries:
-            _message += f", LiteLLM Max Retries: {self.max_retries}"
-        return _message
+        return self._transform_error_to_string()
 
     def __repr__(self):
+        return self._transform_error_to_string()
+
+    def _transform_error_to_string(self) -> str:
+        """
+        Transform the error to a string
+        """
         _message = self.message
         if self.num_retries:
             _message += f" LiteLLM Retried: {self.num_retries} times"

litellm/litellm_core_utils/exception_mapping_utils.py

@@ -74,6 +74,24 @@ class ExceptionCheckers:
             if substring 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.
+        """
+        known_exception_substrings = [
+            "invalid_request_error",
+            "content_policy_violation",
+            "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",
+        ]
+        for substring in known_exception_substrings:
+            if substring in error_str.lower():
+                return True
+        return False
 
 
 def get_error_message(error_obj) -> Optional[str]:
@@ -2021,26 +2039,19 @@ def exception_type(  # type: ignore  # noqa: PLR0915
                         response=getattr(original_exception, "response", None),
                     )
                 elif (
-                    (
-                        "invalid_request_error" in error_str
-                        and "content_policy_violation" in error_str
-                    )
-                    or (
-                        "The response was filtered due to the prompt triggering Azure OpenAI's content management"
-                        in error_str
-                    )
-                    or "Your task failed as a result of our safety system" in error_str
-                    or "The model produced invalid content" in error_str
-                    or "content_filter_policy" in error_str
+                    ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
                 ):
                     exception_mapping_worked = True
-                    raise ContentPolicyViolationError(
-                        message=f"litellm.ContentPolicyViolationError: AzureException - {message}",
-                        llm_provider="azure",
-                        model=model,
-                        litellm_debug_info=extra_information,
-                        response=getattr(original_exception, "response", None),
+                    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 "invalid_request_error" in error_str:
                     exception_mapping_worked = True
                     raise BadRequestError(

litellm/llms/azure/exception_mapping.py

@@ -0,0 +1,42 @@
+from typing import Optional
+
+from litellm.exceptions import ContentPolicyViolationError
+
+
+class AzureOpenAIExceptionMapping:
+    """
+    Class for creating Azure OpenAI specific exceptions
+    """
+    @staticmethod
+    def create_content_policy_violation_error(
+        message: str,
+        model: str,
+        extra_information: str,
+        original_exception: Exception,
+    ) -> ContentPolicyViolationError:
+        """
+        Create a content policy violation error
+        """    
+        raise ContentPolicyViolationError(
+            message=f"litellm.ContentPolicyViolationError: AzureException - {message}",
+            llm_provider="azure",
+            model=model,
+            litellm_debug_info=extra_information,
+            response=getattr(original_exception, "response", None),
+            provider_specific_fields={
+                "innererror": AzureOpenAIExceptionMapping._get_innererror_from_exception(original_exception)
+            },
+        )
+    
+    @staticmethod
+    def _get_innererror_from_exception(original_exception: Exception) -> Optional[dict]:
+        """
+        Azure OpenAI returns the innererror in the body of the exception
+        This method extracts the innererror from the exception
+        """
+        innererror = None
+        body_dict = getattr(original_exception, "body", None) or {}
+        if isinstance(body_dict, dict):
+            innererror = body_dict.get("innererror")
+        return innererror
+            

litellm/proxy/_types.py

@@ -12,6 +12,7 @@ from pydantic import (
     field_validator,
     model_validator,
 )
+from pydantic.v1.types import OptionalInt
 from typing_extensions import Required, TypedDict
 
 from litellm._uuid import uuid
@@ -2669,6 +2670,7 @@ class ProxyException(Exception):
         code: Optional[Union[int, str]] = None,  # maps to status code
         headers: Optional[Dict[str, str]] = None,
         openai_code: Optional[str] = None,  # maps to 'code'  in openai
+        provider_specific_fields: Optional[dict] = None,
     ):
         self.message = str(message)
         self.type = type
@@ -2683,7 +2685,7 @@ class ProxyException(Exception):
                 if not isinstance(v, str):
                     headers[k] = str(v)
         self.headers = headers or {}
-
+        self.provider_specific_fields = provider_specific_fields
         # rules for proxyExceptions
         # Litellm router.py returns "No healthy deployment available" when there are no deployments available
         # Should map to 429 errors https://github.com/BerriAI/litellm/issues/2487
@@ -2697,12 +2699,15 @@ class ProxyException(Exception):
 
     def to_dict(self) -> dict:
         """Converts the ProxyException instance to a dictionary."""
-        return {
+        error_dict: Dict[str, Optional[Union[str, Dict]]] = {
             "message": self.message,
             "type": self.type,
             "param": self.param,
             "code": self.code,
         }
+        if self.provider_specific_fields:
+            error_dict["provider_specific_fields"] = self.provider_specific_fields
+        return error_dict
 
 
 class CommonProxyErrors(str, enum.Enum):

litellm/proxy/common_request_processing.py

@@ -746,6 +746,7 @@ class ProxyBaseLLMRequestProcessing:
                 type=getattr(e, "type", "None"),
                 param=getattr(e, "param", "None"),
                 code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST),
+                provider_specific_fields=getattr(e, "provider_specific_fields", None),
                 headers=headers,
             )
         elif isinstance(e, httpx.HTTPStatusError):
@@ -765,6 +766,7 @@ class ProxyBaseLLMRequestProcessing:
             param=getattr(e, "param", "None"),
             openai_code=getattr(e, "code", None),
             code=getattr(e, "status_code", 500),
+            provider_specific_fields=getattr(e, "provider_specific_fields", None),
             headers=headers,
         )
 

litellm/proxy/proxy_server.py

@@ -877,18 +877,12 @@ class UserAPIKeyCacheTTLEnum(enum.Enum):
 async def openai_exception_handler(request: Request, exc: ProxyException):
     # NOTE: DO NOT MODIFY THIS, its crucial to map to Openai exceptions
     headers = exc.headers
+    error_dict = exc.to_dict()
     return JSONResponse(
         status_code=(
             int(exc.code) if exc.code else status.HTTP_500_INTERNAL_SERVER_ERROR
         ),
-        content={
-            "error": {
-                "message": exc.message,
-                "type": exc.type,
-                "param": exc.param,
-                "code": exc.code,
-            }
-        },
+        content={"error": error_dict},
         headers=headers,
     )
 

tests/llm_translation/test_azure_openai.py

@@ -648,4 +648,61 @@ def test_completion_azure_deployment_id():
         ],
     )
     # Add any assertions here to check the response
-    print(response)
+    print(response)
+def test_azure_with_content_safety_error():
+    """
+    Verify user can access innererror from the Azure OpenAI exception
+    """
+    from litellm import completion
+    from litellm.exceptions import ContentPolicyViolationError
+    from litellm.litellm_core_utils.exception_mapping_utils import exception_type
+    from unittest.mock import MagicMock
+    
+    mock_exception = Exception("The response was filtered due to the prompt triggering Azure OpenAI's content management policy")
+    mock_exception.body = {
+        "innererror": {
+            "code": "ResponsibleAIPolicyViolation",
+            "content_filter_result": {
+                "hate": {
+                    "filtered": False,
+                    "severity": "safe"
+                },
+                "jailbreak": {
+                    "filtered": False,
+                    "detected": False
+                },
+                "self_harm": {
+                    "filtered": False,
+                    "severity": "safe"
+                },
+                "sexual": {
+                    "filtered": False,
+                    "severity": "safe"
+                },
+                "violence": {
+                    "filtered": True,
+                    "severity": "high"
+                }
+            }
+        }
+    }
+    
+    mock_response = MagicMock()
+    mock_response.status_code = 400
+    mock_exception.response = mock_response
+    
+    with pytest.raises(ContentPolicyViolationError) as exc_info:
+        exception_type(
+            model="azure/gpt-4o-new-test",
+            original_exception=mock_exception,
+            custom_llm_provider="azure"
+        )
+    
+    e = exc_info.value
+    print("got exception=", e)
+    assert e.provider_specific_fields is not None
+    print("got provider_specific_fields=", e.provider_specific_fields)
+    assert e.provider_specific_fields.get("innererror") is not None
+    assert e.provider_specific_fields["innererror"]["code"] == "ResponsibleAIPolicyViolation"
+    assert e.provider_specific_fields["innererror"]["content_filter_result"]["violence"]["filtered"] is True
+    assert e.provider_specific_fields["innererror"]["content_filter_result"]["violence"]["severity"] == "high"

tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py

@@ -1,8 +1,18 @@
+import os
+import sys
+
 import pytest
+
 import litellm
 
-from litellm.litellm_core_utils.exception_mapping_utils import ExceptionCheckers
-from litellm.litellm_core_utils.exception_mapping_utils import exception_type
+sys.path.insert(
+    0, os.path.abspath("../../..")
+)  # Adds the parent directory to the system path
+
+from litellm.litellm_core_utils.exception_mapping_utils import (
+    ExceptionCheckers,
+    exception_type,
+)
 
 # Test cases for is_error_str_context_window_exceeded
 # Tuple format: (error_message, expected_result)
@@ -44,6 +54,82 @@ def test_is_error_str_context_window_exceeded(error_str, expected):
     """
     assert ExceptionCheckers.is_error_str_context_window_exceeded(error_str) == expected
 
+class TestExceptionCheckers:
+    """Test the ExceptionCheckers utility methods"""
+
+    def test_is_azure_content_policy_violation_error_with_policy_violation_text(self):
+        """Test detection of Azure content policy violation with explicit policy violation text"""
+        
+        error_strings = [
+            "invalid_request_error content_policy_violation occurred",
+            "The response was filtered due to the prompt triggering Azure OpenAI's content management policy",
+            "Your task failed as a result of our safety system detecting harmful content",
+            "The model produced invalid content that violates our policy",
+            "Request blocked due to content_filter_policy restrictions"
+        ]
+        
+        for error_str in error_strings:
+            result = ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
+            assert result is True, f"Should detect policy violation in: {error_str}"
+
+    def test_is_azure_content_policy_violation_error_case_insensitive(self):
+        """Test that content policy violation detection is case insensitive"""
+        
+        error_strings = [
+            "INVALID_REQUEST_ERROR CONTENT_POLICY_VIOLATION",
+            "The Response Was Filtered Due To The Prompt Triggering Azure OpenAI's Content Management",
+            "YOUR TASK FAILED AS A RESULT OF OUR SAFETY SYSTEM",
+            "Content_Filter_Policy restriction detected"
+        ]
+        
+        for error_str in error_strings:
+            result = ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
+            assert result is True, f"Should detect policy violation in uppercase: {error_str}"
+
+    def test_is_azure_content_policy_violation_error_with_non_policy_errors(self):
+        """Test that non-policy violation errors are not detected as policy violations"""
+        
+        error_strings = [
+            "Invalid API key provided",
+            "Rate limit exceeded for current model",
+            "Model not found: gpt-nonexistent",
+            "Request timeout occurred",
+            "Authentication failed",
+            "Insufficient quota remaining",
+            "Bad request format",
+            "Internal server error occurred"
+        ]
+        
+        for error_str in error_strings:
+            result = ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
+            assert result is False, f"Should NOT detect policy violation in: {error_str}"
+
+    def test_is_azure_content_policy_violation_error_with_partial_matches(self):
+        """Test that partial keyword matches work correctly"""
+        
+        # These should match because they contain the required substrings
+        positive_cases = [
+            "Error: content_policy_violation detected in request",
+            "Safety content management, your task failed as a result of our safety system",
+            "the model produced invalid content",
+        ]
+        
+        for error_str in positive_cases:
+            print("testing positive case=", error_str)
+            result = ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
+            assert result is True, f"Should detect policy violation in: {error_str}"
+        
+        # These should not match even though they contain similar words
+        negative_cases = [
+            "Invalid content format in request",  # "invalid" but not "invalid content"
+            "Policy configuration error",         # "policy" but not policy violation context
+            "Content type not supported",         # "content" but not content filter context  
+            "Management API unavailable"          # "management" but not content management context
+        ]
+        
+        for error_str in negative_cases:
+            result = ExceptionCheckers.is_azure_content_policy_violation_error(error_str)
+            assert result is False, f"Should NOT detect policy violation in: {error_str}"
 
 # Test cases for Vertex AI RateLimitError mapping
 # As per https://github.com/BerriAI/litellm/issues/16189

tests/test_litellm/llms/azure/test_azure_exception_mapping.py

@@ -0,0 +1,193 @@
+import json
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(
+    0, os.path.abspath("../../../..")
+)  # Adds the parent directory to the system path
+
+import litellm
+from litellm.exceptions import ContentPolicyViolationError
+from litellm.litellm_core_utils.exception_mapping_utils import exception_type
+
+
+class TestAzureExceptionMapping:
+    """Test Azure OpenAI exception mapping with provider-specific fields"""
+
+    def test_azure_content_policy_violation_innererror_access(self):
+        """Test that Azure content policy violation exceptions provide access to innererror details"""
+        
+        # Create a mock Azure OpenAI exception with body containing innererror
+        mock_exception = Exception("The response was filtered due to the prompt triggering Azure OpenAI's content management policy")
+        mock_exception.body = {
+            "innererror": {
+                "code": "ResponsibleAIPolicyViolation",
+                "content_filter_result": {
+                    "hate": {
+                        "filtered": True,
+                        "severity": "high"
+                    },
+                    "jailbreak": {
+                        "filtered": False,
+                        "detected": False
+                    },
+                    "self_harm": {
+                        "filtered": False,
+                        "severity": "safe"
+                    },
+                    "sexual": {
+                        "filtered": False,
+                        "severity": "safe"
+                    },
+                    "violence": {
+                        "filtered": True,
+                        "severity": "medium"
+                    }
+                }
+            }
+        }
+        
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_exception.response = mock_response
+        
+        # Test the exception mapping directly
+        with pytest.raises(ContentPolicyViolationError) as exc_info:
+            exception_type(
+                model="azure/gpt-4",
+                original_exception=mock_exception,
+                custom_llm_provider="azure"
+            )
+        
+        # Access the exception and verify provider_specific_fields
+        e = exc_info.value
+        assert e.provider_specific_fields is not None
+        assert "innererror" in e.provider_specific_fields
+        
+        innererror = e.provider_specific_fields["innererror"]
+        assert innererror["code"] == "ResponsibleAIPolicyViolation"
+        assert "content_filter_result" in innererror
+        
+        content_filter_result = innererror["content_filter_result"]
+        assert content_filter_result["hate"]["filtered"] is True
+        assert content_filter_result["hate"]["severity"] == "high"
+        assert content_filter_result["violence"]["filtered"] is True
+        assert content_filter_result["violence"]["severity"] == "medium"
+        assert content_filter_result["sexual"]["filtered"] is False
+        assert content_filter_result["self_harm"]["filtered"] is False
+        assert content_filter_result["jailbreak"]["filtered"] is False
+
+    def test_azure_content_policy_violation_different_categories(self):
+        """Test Azure content policy violation with different filtering categories"""
+        
+        # Mock exception with different content filter results  
+        mock_exception = Exception("The response was filtered due to the prompt triggering Azure OpenAI's content management policy")
+        mock_exception.body = {
+            "innererror": {
+                "code": "ResponsibleAIPolicyViolation",
+                "content_filter_result": {
+                    "hate": {
+                        "filtered": False,
+                        "severity": "safe"
+                    },
+                    "jailbreak": {
+                        "filtered": True,
+                        "detected": True
+                    },
+                    "self_harm": {
+                        "filtered": True,
+                        "severity": "high"
+                    },
+                    "sexual": {
+                        "filtered": True,
+                        "severity": "medium"
+                    },
+                    "violence": {
+                        "filtered": False,
+                        "severity": "safe"
+                    }
+                }
+            }
+        }
+        
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_exception.response = mock_response
+        
+        # Test the exception mapping directly with different violation type
+        with pytest.raises(ContentPolicyViolationError) as exc_info:
+            exception_type(
+                model="azure/gpt-4",
+                original_exception=mock_exception,
+                custom_llm_provider="azure"
+            )
+        
+        # Verify provider_specific_fields contains the expected innererror structure
+        e = exc_info.value
+        assert e.provider_specific_fields is not None
+        print("got provider_specific_fields=", e.provider_specific_fields)
+        innererror = e.provider_specific_fields["innererror"]
+        content_filter_result = innererror["content_filter_result"]
+        
+        # Check different filter categories
+        assert content_filter_result["sexual"]["filtered"] is True
+        assert content_filter_result["sexual"]["severity"] == "medium"
+        assert content_filter_result["self_harm"]["filtered"] is True
+        assert content_filter_result["self_harm"]["severity"] == "high" 
+        assert content_filter_result["jailbreak"]["filtered"] is True
+        assert content_filter_result["jailbreak"]["detected"] is True
+        assert content_filter_result["hate"]["filtered"] is False
+        assert content_filter_result["violence"]["filtered"] is False
+
+    def test_azure_content_policy_violation_missing_innererror(self):
+        """Test Azure content policy violation when innererror is missing from response"""
+        
+        # Mock exception without body attribute
+        mock_exception = Exception("The response was filtered due to the prompt triggering Azure OpenAI's content management policy")
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_exception.response = mock_response
+        # Note: no mock_exception.body attribute set
+        
+        # Test the exception mapping directly
+        with pytest.raises(ContentPolicyViolationError) as exc_info:
+            exception_type(
+                model="azure/gpt-4",
+                original_exception=mock_exception,
+                custom_llm_provider="azure"
+            )
+        
+        # Verify that even without innererror, the exception is still raised properly
+        e = exc_info.value
+        print("got exception=", e)
+        # provider_specific_fields should still exist but innererror should be None
+        assert e.provider_specific_fields is not None
+        assert e.provider_specific_fields.get("innererror") is None
+
+    def test_azure_content_policy_violation_non_dict_body(self):
+        """Test Azure content policy violation when body is not a dictionary"""
+        
+        # Mock exception with non-dict body
+        mock_exception = Exception("The response was filtered due to the prompt triggering Azure OpenAI's content management policy")
+        mock_exception.body = "invalid body format"
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_exception.response = mock_response
+        
+        # Test the exception mapping directly
+        with pytest.raises(ContentPolicyViolationError) as exc_info:
+            exception_type(
+                model="azure/gpt-4",
+                original_exception=mock_exception,
+                custom_llm_provider="azure"
+            )
+        
+        # Verify that with invalid body format, innererror should be None
+        e = exc_info.value
+        print("got exception=", e)
+        print("exception fields=", vars(e))
+        assert e.provider_specific_fields is not None
+        assert e.provider_specific_fields.get("innererror") is None