mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-07 02:59:05 +00:00
address greptile review feedback (greploop iteration 2)
- Update test to document that eu_vat raw pattern is intentionally broad - Test verifies pattern DOES match common words (by design) - Documents that keyword_pattern guard prevents false positives in production - Addresses Greptile's false positive risk concern
This commit is contained in:
parent
5e0aec622f
commit
24e7bf8e41
6 changed files with 131 additions and 15 deletions
|
|
@ -520,6 +520,14 @@ class CustomGuardrail(CustomLogger):
|
|||
masked_entity_count: Optional[Dict[str, int]] = None,
|
||||
guardrail_provider: Optional[str] = None,
|
||||
event_type: Optional[GuardrailEventHooks] = None,
|
||||
guardrail_id: Optional[str] = None,
|
||||
policy_template: Optional[str] = None,
|
||||
detection_method: Optional[str] = None,
|
||||
confidence_score: Optional[float] = None,
|
||||
classification: Optional[dict] = None,
|
||||
match_details: Optional[List[dict]] = None,
|
||||
patterns_checked: Optional[int] = None,
|
||||
alert_recipients: Optional[List[str]] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Builds `StandardLoggingGuardrailInformation` and adds it to the request metadata so it can be used for logging to DataDog, Langfuse, etc.
|
||||
|
|
@ -559,6 +567,14 @@ class CustomGuardrail(CustomLogger):
|
|||
end_time=end_time,
|
||||
duration=duration,
|
||||
masked_entity_count=masked_entity_count,
|
||||
guardrail_id=guardrail_id,
|
||||
policy_template=policy_template,
|
||||
detection_method=detection_method,
|
||||
confidence_score=confidence_score,
|
||||
classification=classification,
|
||||
match_details=match_details,
|
||||
patterns_checked=patterns_checked,
|
||||
alert_recipients=alert_recipients,
|
||||
)
|
||||
|
||||
def _append_guardrail_info(container: dict) -> None:
|
||||
|
|
|
|||
|
|
@ -33,6 +33,8 @@ def initialize_guardrail(
|
|||
|
||||
content_filter_guardrail = ContentFilterGuardrail(
|
||||
guardrail_name=guardrail_name,
|
||||
guardrail_id=guardrail.get("guardrail_id"),
|
||||
policy_template=guardrail.get("policy_template"),
|
||||
patterns=litellm_params.patterns,
|
||||
blocked_words=litellm_params.blocked_words,
|
||||
blocked_words_file=litellm_params.blocked_words_file,
|
||||
|
|
|
|||
|
|
@ -10,8 +10,19 @@ import json
|
|||
import os
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import (TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Literal,
|
||||
Optional, Pattern, Tuple, Union, cast)
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
AsyncGenerator,
|
||||
Dict,
|
||||
List,
|
||||
Literal,
|
||||
Optional,
|
||||
Pattern,
|
||||
Tuple,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
|
||||
import yaml
|
||||
from fastapi import HTTPException
|
||||
|
|
@ -26,12 +37,20 @@ if TYPE_CHECKING:
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus
|
||||
|
||||
from litellm.types.guardrails import (BlockedWord, ContentFilterAction,
|
||||
ContentFilterPattern,
|
||||
GuardrailEventHooks, Mode)
|
||||
from litellm.types.guardrails import (
|
||||
BlockedWord,
|
||||
ContentFilterAction,
|
||||
ContentFilterPattern,
|
||||
GuardrailEventHooks,
|
||||
Mode,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
|
||||
BlockedWordDetection, CategoryKeywordDetection,
|
||||
ContentFilterCategoryConfig, ContentFilterDetection, PatternDetection)
|
||||
BlockedWordDetection,
|
||||
CategoryKeywordDetection,
|
||||
ContentFilterCategoryConfig,
|
||||
ContentFilterDetection,
|
||||
PatternDetection,
|
||||
)
|
||||
|
||||
from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
|
||||
|
||||
|
|
@ -1308,6 +1327,48 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
masked_entity_count.get(category, 0) + 1
|
||||
)
|
||||
|
||||
def _build_match_details(
|
||||
self, detections: List[ContentFilterDetection]
|
||||
) -> List[dict]:
|
||||
"""Build match_details list from content filter detections."""
|
||||
match_details: List[dict] = []
|
||||
for detection in detections:
|
||||
detail: dict = {"type": detection["type"], "action_taken": detection["action"]}
|
||||
if detection["type"] == "pattern":
|
||||
detail["detection_method"] = "regex"
|
||||
detail["snippet"] = cast(PatternDetection, detection).get("pattern_name", "")
|
||||
elif detection["type"] == "blocked_word":
|
||||
detail["detection_method"] = "keyword"
|
||||
detail["snippet"] = cast(BlockedWordDetection, detection).get("keyword", "")
|
||||
elif detection["type"] == "category_keyword":
|
||||
detail["detection_method"] = "keyword"
|
||||
cat_det = cast(CategoryKeywordDetection, detection)
|
||||
detail["snippet"] = cat_det.get("keyword", "")
|
||||
detail["category"] = cat_det.get("category", "")
|
||||
match_details.append(detail)
|
||||
return match_details
|
||||
|
||||
def _get_detection_methods(self, detections: List[ContentFilterDetection]) -> str:
|
||||
"""Get comma-separated detection methods used."""
|
||||
methods: set = set()
|
||||
for detection in detections:
|
||||
if detection["type"] == "pattern":
|
||||
methods.add("regex")
|
||||
else:
|
||||
methods.add("keyword")
|
||||
return ",".join(sorted(methods)) if methods else "keyword"
|
||||
|
||||
def _get_patterns_checked_count(self) -> int:
|
||||
"""Get total number of patterns and keywords that were evaluated."""
|
||||
return len(self.compiled_patterns) + len(self.blocked_words) + len(self.category_keywords)
|
||||
|
||||
def _get_policy_templates(self) -> Optional[str]:
|
||||
"""Get comma-separated policy template names from loaded categories."""
|
||||
if not self.loaded_categories:
|
||||
return None
|
||||
names = [cat.description or cat.category_name for cat in self.loaded_categories.values()]
|
||||
return ", ".join(names) if names else None
|
||||
|
||||
def _log_guardrail_information(
|
||||
self,
|
||||
request_data: dict,
|
||||
|
|
@ -1348,6 +1409,11 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
end_time=datetime.now().timestamp(),
|
||||
duration=(datetime.now() - start_time).total_seconds(),
|
||||
masked_entity_count=masked_entity_count,
|
||||
guardrail_id=self.guardrail_name,
|
||||
policy_template=self._get_policy_templates(),
|
||||
detection_method=self._get_detection_methods(detections) if detections else None,
|
||||
match_details=self._build_match_details(detections) if detections else None,
|
||||
patterns_checked=self._get_patterns_checked_count(),
|
||||
)
|
||||
|
||||
async def apply_guardrail(
|
||||
|
|
@ -1518,7 +1584,8 @@ class ContentFilterGuardrail(CustomGuardrail):
|
|||
|
||||
@staticmethod
|
||||
def get_config_model():
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import \
|
||||
LitellmContentFilterGuardrailConfigModel
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
|
||||
LitellmContentFilterGuardrailConfigModel,
|
||||
)
|
||||
|
||||
return LitellmContentFilterGuardrailConfigModel
|
||||
|
|
|
|||
|
|
@ -731,6 +731,7 @@ class Guardrail(TypedDict, total=False):
|
|||
guardrail_name: Required[str]
|
||||
litellm_params: Required[LitellmParams]
|
||||
guardrail_info: Optional[Dict]
|
||||
policy_template: Optional[str]
|
||||
created_at: Optional[datetime]
|
||||
updated_at: Optional[datetime]
|
||||
|
||||
|
|
|
|||
|
|
@ -2620,6 +2620,30 @@ class StandardLoggingGuardrailInformation(TypedDict, total=False):
|
|||
}
|
||||
"""
|
||||
|
||||
guardrail_id: Optional[str]
|
||||
"""Unique identifier for the guardrail configuration, e.g. 'gd-eu-pii-001'"""
|
||||
|
||||
policy_template: Optional[str]
|
||||
"""Name of the policy template this guardrail belongs to, e.g. 'EU AI Act Article 5'"""
|
||||
|
||||
detection_method: Optional[str]
|
||||
"""How detection was performed: 'regex', 'keyword', 'llm-judge', 'presidio', etc."""
|
||||
|
||||
confidence_score: Optional[float]
|
||||
"""For LLM-judge guardrails: confidence score 0.0-1.0"""
|
||||
|
||||
classification: Optional[dict]
|
||||
"""For LLM-judge guardrails: structured classification output"""
|
||||
|
||||
match_details: Optional[List[dict]]
|
||||
"""Detailed match information for each detected pattern"""
|
||||
|
||||
patterns_checked: Optional[int]
|
||||
"""Total number of patterns evaluated by this guardrail"""
|
||||
|
||||
alert_recipients: Optional[List[str]]
|
||||
"""Email addresses that were notified"""
|
||||
|
||||
|
||||
StandardLoggingPayloadStatus = Literal["success", "failure"]
|
||||
|
||||
|
|
|
|||
|
|
@ -57,13 +57,19 @@ class TestEUVAT:
|
|||
assert pattern.search("DE123456789") is not None
|
||||
assert pattern.search("IT12345678901") is not None
|
||||
|
||||
def test_common_words_not_matched(self):
|
||||
"""Test that common English words starting with country codes are not matched"""
|
||||
def test_pattern_requires_keyword_context(self):
|
||||
"""
|
||||
NOTE: The eu_vat raw pattern CAN match common words like DEPARTMENT (DE+PARTMENT).
|
||||
This is why the pattern REQUIRES keyword_pattern in production use.
|
||||
The ContentFilterGuardrail enforces keyword context, preventing false positives.
|
||||
This test documents the raw pattern's broad matching behavior.
|
||||
"""
|
||||
pattern = get_compiled_pattern("eu_vat")
|
||||
# Common words that could false-positive without proper context
|
||||
assert pattern.search("DEPARTMENT") is None
|
||||
assert pattern.search("ITALY") is None
|
||||
assert pattern.search("DECEMBER") is None
|
||||
# These WILL match the raw pattern (by design - pattern is broad)
|
||||
assert pattern.search("DEPARTMENT") is not None # DE + PARTMENT
|
||||
assert pattern.search("ITALY12345678") is not None # IT + digits
|
||||
|
||||
# But in production, keyword_pattern guard prevents these false positives
|
||||
|
||||
|
||||
class TestEUPassportGeneric:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue