fix(presidio): handle empty content and error dict responses (#17489)

- Skip empty/whitespace text before calling Presidio API
- Handle error dict responses gracefully (e.g., {'error': 'No text provided'})
- Add defensive error handling for invalid result items
- Add comprehensive test coverage for empty content scenarios

Fixes crash in tool/function calling where assistant messages have empty content.
This commit is contained in:
Dominic Fallows 2025-12-05 23:45:19 +00:00 committed by GitHub
parent 5fb7530d8c
commit 2ffe8ee204
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 264 additions and 1 deletions

View file

@ -207,6 +207,14 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
Send text to the Presidio analyzer endpoint and get analysis results
"""
try:
# Skip empty or whitespace-only text to avoid Presidio errors
# Common in tool/function calling where assistant content is empty
if not text or len(text.strip()) == 0:
verbose_proxy_logger.debug(
"Skipping Presidio analysis for empty/whitespace-only text"
)
return []
async with aiohttp.ClientSession() as session:
if self.mock_redacted_text is not None:
return self.mock_redacted_text
@ -231,9 +239,42 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
async with session.post(analyze_url, json=analyze_payload) as response:
analyze_results = await response.json()
verbose_proxy_logger.debug("analyze_results: %s", analyze_results)
# Handle error responses from Presidio (e.g., {'error': 'No text provided'})
# Presidio may return a dict instead of a list when errors occur
if isinstance(analyze_results, dict):
if "error" in analyze_results:
verbose_proxy_logger.warning(
"Presidio analyzer returned error: %s, returning empty list",
analyze_results.get("error")
)
return []
# If it's a dict but not an error, try to process it as a single item
verbose_proxy_logger.debug(
"Presidio returned dict (not list), attempting to process as single item"
)
try:
return [PresidioAnalyzeResponseItem(**analyze_results)]
except Exception as e:
verbose_proxy_logger.warning(
"Failed to parse Presidio dict response: %s, returning empty list",
e
)
return []
# Normal case: list of results
final_results = []
for item in analyze_results:
final_results.append(PresidioAnalyzeResponseItem(**item))
try:
final_results.append(PresidioAnalyzeResponseItem(**item))
except TypeError as te:
# Handle case where item is not a dict (shouldn't happen, but be defensive)
verbose_proxy_logger.warning(
"Skipping invalid Presidio result item: %s (error: %s)",
item,
te
)
continue
return final_results
except Exception as e:
raise e

View file

@ -634,6 +634,228 @@ async def test_request_data_flows_to_apply_guardrail():
print("✓ request_data correctly passed to apply_guardrail")
@pytest.mark.asyncio
async def test_empty_content_handling(presidio_guardrail, mock_user_api_key, mock_cache):
"""
Test that Presidio handles empty content gracefully.
This is common in tool/function calling where assistant messages have
empty content but include tool_calls.
Bug fix: Previously crashed with:
TypeError: argument after ** must be a mapping, not str
"""
test_data = {
"messages": [
{"role": "user", "content": "What is 2+2?"},
{
"role": "assistant",
"content": "", # Empty content - common in tool calls
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {"name": "calculator", "arguments": '{"a":2,"b":2}'},
}
],
},
{"role": "tool", "tool_call_id": "call_123", "content": "4"},
],
"model": "gpt-4",
}
# Mock check_pii to simulate PII processing without needing Presidio API
async def mock_check_pii(text, output_parse_pii, presidio_config, request_data):
# Empty text returns as-is (this is what our fix ensures)
return text
presidio_guardrail.check_pii = mock_check_pii
# This should not raise an exception
result = await presidio_guardrail.async_pre_call_hook(
user_api_key_dict=mock_user_api_key,
cache=mock_cache,
data=test_data,
call_type="completion",
)
assert result is not None
assert "messages" in result
# Verify messages are preserved
assert len(result["messages"]) == 3
print("✓ Empty content handling test passed")
@pytest.mark.asyncio
async def test_whitespace_only_content(presidio_guardrail, mock_user_api_key, mock_cache):
"""
Test that Presidio handles whitespace-only content gracefully.
Whitespace-only content should be treated the same as empty content.
"""
test_data = {
"messages": [
{"role": "user", "content": " "}, # Whitespace only
{"role": "assistant", "content": "\n\t "}, # Tabs and newlines
{"role": "user", "content": "Real question here"},
],
"model": "gpt-4",
}
# Mock check_pii to simulate PII processing
async def mock_check_pii(text, output_parse_pii, presidio_config, request_data):
return text
presidio_guardrail.check_pii = mock_check_pii
result = await presidio_guardrail.async_pre_call_hook(
user_api_key_dict=mock_user_api_key,
cache=mock_cache,
data=test_data,
call_type="completion",
)
assert result is not None
assert len(result["messages"]) == 3
print("✓ Whitespace-only content test passed")
@pytest.mark.asyncio
async def test_analyze_text_with_empty_string():
"""
Test analyze_text method directly with empty string.
Should return empty list without making API call to Presidio.
"""
presidio = _OPTIONAL_PresidioPIIMasking(
presidio_analyzer_api_base="http://test:5002/",
presidio_anonymizer_api_base="http://test:5001/",
output_parse_pii=False,
)
# Test with empty string - should return immediately without API call
result = await presidio.analyze_text(
text="",
presidio_config=None,
request_data={},
)
assert result == [], "Empty text should return empty list"
# Test with whitespace only - should return immediately
result = await presidio.analyze_text(
text=" \n\t ",
presidio_config=None,
request_data={},
)
assert result == [], "Whitespace-only text should return empty list"
print("✓ analyze_text empty string test passed")
@pytest.mark.asyncio
async def test_analyze_text_error_dict_handling():
"""
Test that analyze_text handles error dict responses from Presidio API.
When Presidio returns {'error': 'No text provided'}, should handle gracefully
instead of crashing with TypeError.
"""
presidio = _OPTIONAL_PresidioPIIMasking(
presidio_analyzer_api_base="http://mock-presidio:5002/",
presidio_anonymizer_api_base="http://mock-presidio:5001/",
output_parse_pii=False,
)
# Mock the HTTP response to return error dict
class MockResponse:
async def json(self):
return {"error": "No text provided"}
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
class MockSession:
def post(self, *args, **kwargs):
return MockResponse()
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
with patch("aiohttp.ClientSession", return_value=MockSession()):
result = await presidio.analyze_text(
text="some text",
presidio_config=None,
request_data={},
)
# Should return empty list when error dict is received
assert result == [], "Error dict should be handled gracefully"
print("✓ analyze_text error dict handling test passed")
@pytest.mark.asyncio
async def test_tool_calling_complete_scenario(presidio_guardrail, mock_user_api_key, mock_cache):
"""
Test complete tool calling scenario with PII in user message.
This tests the real-world scenario where:
1. User provides a query with PII
2. Assistant responds with empty content + tool_calls
3. Tool provides response
4. Assistant provides final answer
"""
test_data = {
"messages": [
{
"role": "user",
"content": "My email is john.doe@example.com. Can you look up my account?",
},
{
"role": "assistant",
"content": "", # Empty - tool call
"tool_calls": [
{
"id": "call_abc",
"type": "function",
"function": {"name": "lookup_account", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_abc", "content": "Account found"},
{"role": "assistant", "content": "I found your account information."},
],
"model": "gpt-4",
}
# Mock check_pii to simulate PII masking
async def mock_check_pii(text, output_parse_pii, presidio_config, request_data):
if "john.doe@example.com" in text:
return text.replace("john.doe@example.com", "[EMAIL]")
return text
presidio_guardrail.check_pii = mock_check_pii
result = await presidio_guardrail.async_pre_call_hook(
user_api_key_dict=mock_user_api_key,
cache=mock_cache,
data=test_data,
call_type="completion",
)
assert result is not None
# Verify PII was masked in user message
assert "[EMAIL]" in result["messages"][0]["content"]
assert "john.doe@example.com" not in result["messages"][0]["content"]
# Verify other messages preserved
assert len(result["messages"]) == 4
print("✓ Tool calling complete scenario test passed")
if __name__ == "__main__":
# Run tests
asyncio.run(