scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-17 23:51:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 67

added bedrock guardrail API exception

Browse source
This commit is contained in:
kothamah 2026-03-19 13:52:23 -04:00 • committed by GitHub
parent e5baa2232f
commit a5dd01d8ac
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 23 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
View file

@ -82,7 +82,12 @@ def _redact_pii_matches(response_json: dict) -> dict:
redacted_response = copy.deepcopy(response_json)
# Get assessments from the response
assessments = redacted_response.get("assessments", [])
# NOTE: We use `.get("key") or []` instead of `.get("key", [])` because
# the Bedrock API can return explicit `null` for list fields (e.g. "regexes": null).
# In Python, dict.get("key", []) returns None (not []) when the key exists
# with a None/null value. The `or []` ensures we always get an iterable,
# preventing "TypeError: 'NoneType' object is not iterable".
assessments = redacted_response.get("assessments") or []
if not assessments:
return redacted_response
@ -90,13 +95,13 @@ def _redact_pii_matches(response_json: dict) -> dict:
# Redact PII entities in sensitive information policy
sensitive_info_policy = assessment.get("sensitiveInformationPolicy")
if sensitive_info_policy:
pii_entities = sensitive_info_policy.get("piiEntities", [])
pii_entities = sensitive_info_policy.get("piiEntities") or []
for pii_entity in pii_entities:
if "match" in pii_entity:
pii_entity["match"] = "[REDACTED]"
# Redact regex matches
regexes = sensitive_info_policy.get("regexes", [])
regexes = sensitive_info_policy.get("regexes") or []
for regex_match in regexes:
if "match" in regex_match:
regex_match["match"] = "[REDACTED]"
@ -104,12 +109,12 @@ def _redact_pii_matches(response_json: dict) -> dict:
# Redact custom word matches in word policy
word_policy = assessment.get("wordPolicy")
if word_policy:
custom_words = word_policy.get("customWords", [])
custom_words = word_policy.get("customWords") or []
for custom_word in custom_words:
if "match" in custom_word:
custom_word["match"] = "[REDACTED]"
managed_words = word_policy.get("managedWordLists", [])
managed_words = word_policy.get("managedWordLists") or []
for managed_word in managed_words:
if "match" in managed_word:
managed_word["match"] = "[REDACTED]"
@ -682,7 +687,9 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
return False
# Check assessments to determine if any actions were BLOCKED (vs ANONYMIZED)
assessments = response.get("assessments", [])
# NOTE: Use `or []` instead of default param to handle explicit null from Bedrock API.
# See _redact_pii_matches() for detailed explanation of the null safety pattern.
assessments = response.get("assessments") or []
if not assessments:
return False
@ -690,7 +697,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
# Check topic policy
topic_policy = assessment.get("topicPolicy")
if topic_policy:
topics = topic_policy.get("topics", [])
topics = topic_policy.get("topics") or []
for topic in topics:
if topic.get("action") == "BLOCKED":
return True
@ -698,7 +705,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
# Check content policy
content_policy = assessment.get("contentPolicy")
if content_policy:
filters = content_policy.get("filters", [])
filters = content_policy.get("filters") or []
for filter_item in filters:
if filter_item.get("action") == "BLOCKED":
return True
@ -706,11 +713,11 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
# Check word policy
word_policy = assessment.get("wordPolicy")
if word_policy:
custom_words = word_policy.get("customWords", [])
custom_words = word_policy.get("customWords") or []
for custom_word in custom_words:
if custom_word.get("action") == "BLOCKED":
return True
managed_words = word_policy.get("managedWordLists", [])
managed_words = word_policy.get("managedWordLists") or []
for managed_word in managed_words:
if managed_word.get("action") == "BLOCKED":
return True
@ -718,12 +725,12 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
# Check sensitive information policy
sensitive_info_policy = assessment.get("sensitiveInformationPolicy")
if sensitive_info_policy:
pii_entities = sensitive_info_policy.get("piiEntities", [])
pii_entities = sensitive_info_policy.get("piiEntities") or []
if pii_entities:
for pii_entity in pii_entities:
if pii_entity.get("action") == "BLOCKED":
return True
regexes = sensitive_info_policy.get("regexes", [])
regexes = sensitive_info_policy.get("regexes") or []
if regexes:
for regex in regexes:
if regex.get("action") == "BLOCKED":
@ -732,7 +739,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
# Check contextual grounding policy
contextual_grounding_policy = assessment.get("contextualGroundingPolicy")
if contextual_grounding_policy:
grounding_filters = contextual_grounding_policy.get("filters", [])
grounding_filters = contextual_grounding_policy.get("filters") or []
for grounding_filter in grounding_filters:
if grounding_filter.get("action") == "BLOCKED":
return True
@ -1391,7 +1398,9 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
Raises:
Exception: If content is blocked by Bedrock guardrail
"""
texts = inputs.get("texts", [])
# NOTE: Use `or []` to handle case where inputs["texts"] is explicitly None.
# dict.get("texts", []) would return None if the key exists with a None value.
texts = inputs.get("texts") or []
try:
verbose_proxy_logger.debug(
f"Bedrock Guardrail: Applying guardrail to {len(texts)} text(s)"