feat: add competitor name blocker guardrail

This commit is contained in:
Krrish Dholakia 2026-02-20 16:33:25 -08:00
parent 33d49e92cb
commit 4349bdaa27
45 changed files with 1256 additions and 27 deletions

View file

@ -24,6 +24,33 @@ guardrails:
guardrail: mcp_end_user_permission
mode: pre_call
default_on: true
- guardrail_name: "airline-competitor-intent"
guardrail_id: "airline-competitor-intent"
litellm_params:
guardrail: litellm_content_filter
mode: pre_call
default_on: false
competitor_intent_config:
brand_self:
- emirates
- ek
competitors:
- qatar airways
- qatar
- etihad
locations:
- qatar
- doha
- doh
competitor_aliases:
qatar airways: [qr, doha airline]
qatar: [qr]
policy:
competitor_comparison: refuse
possible_competitor_comparison: reframe
threshold_high: 0.70
threshold_medium: 0.45
threshold_low: 0.30
mcp_servers:
my_http_server:

View file

@ -1,9 +1,8 @@
from typing import TYPE_CHECKING, Optional
import litellm
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
ContentFilterGuardrail,
)
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import \
ContentFilterGuardrail
from litellm.types.guardrails import SupportedGuardrailIntegrations
if TYPE_CHECKING:
@ -44,6 +43,9 @@ def initialize_guardrail(
severity_threshold=getattr(litellm_params, "severity_threshold", "medium"),
llm_router=llm_router,
image_model=getattr(litellm_params, "image_model", None),
competitor_intent_config=getattr(
litellm_params, "competitor_intent_config", None
),
)
litellm.logging_callback_manager.add_litellm_callback(content_filter_guardrail)

View file

@ -0,0 +1,17 @@
"""
Competitor intent: entity + intent disambiguation with safe (non-competitor) defaults.
Base logic in base.py; industry-specific checkers in submodules (e.g. airline.py).
"""
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.competitor_intent.airline import \
AirlineCompetitorIntentChecker
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.competitor_intent.base import (
BaseCompetitorIntentChecker, normalize, text_for_entity_matching)
__all__ = [
"BaseCompetitorIntentChecker",
"AirlineCompetitorIntentChecker",
"normalize",
"text_for_entity_matching",
]

View file

@ -0,0 +1,144 @@
"""
Airline-specific competitor intent: other meaning (e.g. location/travel context) vs competitor airline.
Uses context-based disambiguation only: no hardcoded place lists. Detects travel-location
language (prepositions, travel verbs, booking/entry nouns) vs airline context (airways,
carrier, lounge, miles, etc.) and scores to decide OTHER_MEANING vs COMPETITOR.
"""
from typing import Any, Dict, List, Tuple
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.competitor_intent.base import (
BaseCompetitorIntentChecker, _compile_marker, _count_signals,
_word_boundary_match)
# Location/travel context: prepositions, travel verbs, booking nouns, entry/geo nouns.
# No place-name list; these patterns detect "destination context" generically.
AIRLINE_OTHER_MEANING_SIGNALS = [
# Travel verb + preposition (e.g. "fly to", "layover in")
r"\b(fly|flying|travel|traveling|going|visit|visiting|transit|layover|stopover)\b.{0,12}\b(to|from|via|in|at|through|into)\b",
# Booking + preposition
r"\bflight(s)?\b.{0,10}\b(to|from|via)\b",
r"\bticket(s)?\b.{0,8}\b(to|for)\b",
r"\bfare(s)?\b.{0,8}\b(to)\b",
# Entry/geo/booking single words
r"\bvisa\b",
r"\bimmigration\b",
r"\bcustoms\b",
r"\bentry\b",
r"\bairport\b",
r"\bterminal\b",
r"\bgate\b",
r"\bdeparture\b",
r"\barrival\b",
r"\bitinerary\b",
r"\bweather\b",
r"\bhotel\b",
r"\bcity\b",
# Prepositions alone (weaker; often near a place)
r"\bto\s+",
r"\bfrom\s+",
r"\bin\s+",
r"\bat\s+",
r"\bvia\s+",
]
# Airline context: carrier/airline language, cabin, loyalty, operations.
# If ambiguous token appears near these → treat as COMPETITOR.
AIRLINE_COMPETITOR_SIGNALS = [
r"\bairways?\b",
r"\bairline\b",
r"\bcarrier\b",
r"\bcabin\s+crew\b",
r"\bflight\s+attendant\b",
r"\bbusiness\s+class\b",
r"\bfirst\s+class\b",
r"\beconomy\b",
r"\blounge\b",
r"\bbaggage\s+allowance\b",
r"\bcheck[- ]?in\b",
r"\bmiles\b",
r"\bloyalty\b",
r"\bstatus\b",
r"\bfrequent\s+flyer\b",
r"\bfleet\b",
r"\baircraft\b",
# Comparison/ranking
r"\bbetter\b",
r"\bbest\b",
r"\bgood\b",
r"\bas\s+good\s+as\b",
r"\bvs\.?\b",
r"\bversus\b",
r"\bcompare\b",
r"\balternative\b",
r"\bcompetitor\b",
# Brand-specific (optional; config can extend)
r"\bqmiles\b",
r"\bprivilege\s+club\b",
]
# Explicit markers: strong override when present.
AIRLINE_EXPLICIT_COMPETITOR_MARKER = r"\b(airways?|airline|carrier)\b"
AIRLINE_EXPLICIT_OTHER_MEANING_MARKER = (
r"\b(fly|travel|going|visit|layover|stopover|transit)\b.{0,12}\b(to|in|via|from)\b.{0,8}\b"
)
class AirlineCompetitorIntentChecker(BaseCompetitorIntentChecker):
"""
Disambiguates other meaning (e.g. country/city/airport) vs competitor airline
(e.g. "Qatar" → country vs Qatar Airways). Overrides _classify_ambiguous
with other_meaning/competitor signals and explicit markers.
"""
def __init__(self, config: Dict[str, Any]) -> None:
merged: Dict[str, Any] = dict(config)
if not merged.get("other_meaning_signals"):
merged["other_meaning_signals"] = AIRLINE_OTHER_MEANING_SIGNALS
if not merged.get("competitor_signals"):
merged["competitor_signals"] = AIRLINE_COMPETITOR_SIGNALS
# Optional: no default place list; config can add other_meaning_anchors for extra patterns
if "other_meaning_anchors" not in merged:
merged["other_meaning_anchors"] = []
if not merged.get("explicit_competitor_marker"):
merged["explicit_competitor_marker"] = AIRLINE_EXPLICIT_COMPETITOR_MARKER
if not merged.get("explicit_other_meaning_marker"):
merged["explicit_other_meaning_marker"] = AIRLINE_EXPLICIT_OTHER_MEANING_MARKER
super().__init__(merged)
self._other_meaning_signals = list(merged.get("other_meaning_signals") or [])
self._competitor_signals = list(merged.get("competitor_signals") or [])
self._other_meaning_anchors = list(merged.get("other_meaning_anchors") or [])
self._explicit_competitor_marker = _compile_marker(
merged.get("explicit_competitor_marker")
)
self._explicit_other_meaning_marker = _compile_marker(
merged.get("explicit_other_meaning_marker")
)
def _classify_ambiguous(self, text: str, token: str) -> Tuple[str, float]:
"""Other meaning vs competitor using airline signals and explicit markers."""
text_lower = text.lower()
if self._explicit_competitor_marker and self._explicit_competitor_marker.search(
text_lower
) and _word_boundary_match(text_lower, token.lower()):
return "COMPETITOR", 0.85
if self._explicit_other_meaning_marker and self._explicit_other_meaning_marker.search(
text_lower
):
return "OTHER_MEANING", 0.85
# Score: location/travel context vs airline context (no place-name list)
other_count = _count_signals(text_lower, self._other_meaning_signals)
if self._other_meaning_anchors:
other_count += _count_signals(text_lower, self._other_meaning_anchors)
comp_count = _count_signals(text_lower, self._competitor_signals)
total = other_count + comp_count
if total == 0:
return "OTHER_MEANING", 0.5
other_ratio = other_count / total
comp_ratio = comp_count / total
if other_ratio >= 0.6:
return "OTHER_MEANING", min(0.9, 0.5 + 0.4 * other_ratio)
if comp_ratio >= 0.6:
return "COMPETITOR", min(0.9, 0.5 + 0.4 * comp_ratio)
return "OTHER_MEANING", 0.5

View file

@ -0,0 +1,241 @@
"""
Generic competitor intent checker: two entity sets and overridable disambiguation.
"""
import re
import unicodedata
from typing import Any, Dict, List, Optional, Pattern, Set, Tuple, cast
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
CompetitorActionHint,
CompetitorIntentEvidenceEntry,
CompetitorIntentResult,
)
ZERO_WIDTH = re.compile(r"[\u200b-\u200d\u2060\ufeff]")
LEET = {"@": "a", "4": "a", "0": "o", "3": "e", "1": "i", "5": "s", "7": "t"}
OTHER_MEANING_DEFAULT_THRESHOLD = 0.65 # Below this → treat as non-competitor (safe default).
def normalize(text: str) -> str:
"""Lowercase, NFKC, strip zero-width, leetspeak, collapse spaces."""
if not text or not isinstance(text, str):
return ""
t = ZERO_WIDTH.sub("", text)
t = unicodedata.normalize("NFKC", t).lower().strip()
for c, r in LEET.items():
t = t.replace(c, r)
return re.sub(r"\s+", " ", t)
def _word_boundary_match(text: str, token: str) -> bool:
"""True if token appears as a word in text."""
return bool(re.search(r"\b" + re.escape(token) + r"\b", text))
def _count_signals(text: str, patterns: List[str]) -> int:
"""Count how many of the patterns appear in text."""
return sum(1 for p in patterns if re.search(p, text, re.IGNORECASE))
def _compile_marker(pattern: Optional[str]) -> Optional[Pattern[str]]:
"""Compile optional regex string to a pattern."""
if not pattern or not pattern.strip():
return None
try:
return re.compile(pattern, re.IGNORECASE)
except re.error:
return None
def text_for_entity_matching(text: str) -> str:
"""Letters-only variant for entity matching (e.g. split punctuation)."""
t = re.sub(r"[^\w\s]", " ", text)
return re.sub(r"\s+", " ", t).strip()
class BaseCompetitorIntentChecker:
"""
Generic competitor intent checker with two entity sets. Ambiguous tokens
(competitor + other-meaning, e.g. location) are classified by overridable
_classify_ambiguous(). Base implementation: treat as non-competitor.
"""
def __init__(self, config: Dict[str, Any]) -> None:
self.brand_self: List[str] = [
s.lower().strip() for s in (config.get("brand_self") or []) if s
]
competitors: List[str] = [
s.lower().strip() for s in (config.get("competitors") or []) if s
]
aliases_map: Dict[str, List[str]] = config.get("competitor_aliases") or {}
self.competitor_canonical: Dict[str, str] = {}
self._competitor_tokens: Set[str] = set()
for c in competitors:
self._competitor_tokens.add(c)
self.competitor_canonical[c] = c
for a in (aliases_map.get(c) or []):
a = a.lower().strip()
if a:
self._competitor_tokens.add(a)
self.competitor_canonical[a] = c
other: List[str] = [
s.lower().strip() for s in (config.get("locations") or []) if s
]
self._other_meaning_tokens: Set[str] = set(other)
self._ambiguous: Set[str] = self._competitor_tokens & self._other_meaning_tokens
self.policy: Dict[str, str] = config.get("policy") or {}
self.threshold_high = float(config.get("threshold_high", 0.70))
self.threshold_medium = float(config.get("threshold_medium", 0.45))
self.threshold_low = float(config.get("threshold_low", 0.30))
self.reframe_message_template: Optional[str] = config.get(
"reframe_message_template"
)
self.refuse_message_template: Optional[str] = config.get(
"refuse_message_template"
)
self._comparison_words: List[str] = list(
config.get("comparison_words")
or ["better", "worse", "best", "vs", "versus", "compare", "alternative", "recommend"]
)
def _classify_ambiguous(self, text: str, token: str) -> Tuple[str, float]:
"""
Override in subclasses for industry-specific logic. Base: treat as non-competitor.
"""
return "OTHER_MEANING", 0.5
def _find_matches(self, text: str) -> List[Tuple[str, str, bool]]:
"""Find competitor matches; mark ambiguous (also in other-meaning set)."""
normalized = normalize(text)
found: List[Tuple[str, str, bool]] = []
seen: Set[Tuple[str, str]] = set()
for token in self._competitor_tokens:
if not _word_boundary_match(normalized, token):
continue
canonical = self.competitor_canonical.get(token, token)
key = (token, canonical)
if key in seen:
continue
seen.add(key)
is_ambig = token in self._ambiguous or token in self._other_meaning_tokens
found.append((token, canonical, is_ambig))
return found
def run(self, text: str) -> CompetitorIntentResult:
"""Classify competitor intent; non-competitor when ambiguous or low confidence."""
normalized = normalize(text)
evidence: List[CompetitorIntentEvidenceEntry] = []
entities: Dict[str, List[str]] = {
"brand_self": [],
"competitors": [],
"category": [],
}
for b in self.brand_self:
if _word_boundary_match(normalized, b):
entities["brand_self"].append(b)
evidence.append(
{"type": "entity", "key": "brand_self", "value": b, "match": b}
)
matches = self._find_matches(text)
if not matches:
return {
"intent": "other",
"confidence": 0.0,
"entities": entities,
"signals": [],
"action_hint": "allow",
"evidence": evidence,
}
competitor_resolved: List[str] = []
for token, canonical, is_ambig in matches:
if not is_ambig:
competitor_resolved.append(canonical)
evidence.append(
{
"type": "entity",
"key": "competitor",
"value": canonical,
"match": token,
}
)
continue
label, conf = self._classify_ambiguous(normalized, token)
if label == "OTHER_MEANING":
evidence.append(
{"type": "signal", "key": "other_meaning", "match": token}
)
continue
if label == "COMPETITOR":
competitor_resolved.append(canonical)
evidence.append(
{
"type": "entity",
"key": "competitor",
"value": canonical,
"match": token,
}
)
if conf < OTHER_MEANING_DEFAULT_THRESHOLD:
competitor_resolved.pop()
evidence.append(
{
"type": "signal",
"key": "other_meaning_default",
"match": f"confidence {conf:.2f}",
}
)
continue
entities["competitors"] = list(dict.fromkeys(competitor_resolved))
if not competitor_resolved:
return {
"intent": "other",
"confidence": 0.0,
"entities": entities,
"signals": ["other_meaning_or_ambiguous"],
"action_hint": "allow",
"evidence": evidence,
}
has_comparison = any(
re.search(r"\b" + re.escape(w) + r"\b", normalized)
for w in self._comparison_words
)
if has_comparison:
evidence.append(
{"type": "signal", "key": "comparison", "match": "comparison language"}
)
confidence = 0.75 if has_comparison else 0.55
if confidence >= self.threshold_high:
intent = "competitor_comparison"
elif confidence >= self.threshold_medium:
intent = "possible_competitor_comparison"
elif confidence >= self.threshold_low:
intent = "log_only"
else:
intent = "other"
action_hint: CompetitorActionHint = cast(
CompetitorActionHint, self.policy.get(intent, "allow")
)
if intent == "log_only":
action_hint = "log_only"
if intent == "other":
action_hint = "allow"
return {
"intent": intent,
"confidence": round(confidence, 2),
"entities": entities,
"signals": ["competitor_resolved"] + (["comparison"] if has_comparison else []),
"action_hint": action_hint,
"evidence": evidence,
}

View file

@ -51,11 +51,14 @@ from litellm.types.guardrails import (
from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
BlockedWordDetection,
CategoryKeywordDetection,
CompetitorIntentDetection,
CompetitorIntentResult,
ContentFilterCategoryConfig,
ContentFilterDetection,
PatternDetection,
)
from .competitor_intent import AirlineCompetitorIntentChecker
from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
MAX_KEYWORD_VALUE_GAP_WORDS = 1
@ -152,6 +155,7 @@ class ContentFilterGuardrail(CustomGuardrail):
severity_threshold: str = "medium",
llm_router: Optional[Router] = None,
image_model: Optional[str] = None,
competitor_intent_config: Optional[Dict[str, Any]] = None,
**kwargs,
):
"""
@ -202,6 +206,22 @@ class ContentFilterGuardrail(CustomGuardrail):
{}
) # category_name -> {identifier_words, block_words, action, severity}
# Competitor intent checker (optional, generic; industry presets add domain_words etc.)
self._competitor_intent_checker: Optional[AirlineCompetitorIntentChecker] = None
if competitor_intent_config and isinstance(competitor_intent_config, dict):
try:
self._competitor_intent_checker = AirlineCompetitorIntentChecker(
competitor_intent_config
)
verbose_proxy_logger.debug(
"ContentFilterGuardrail: competitor intent checker enabled"
)
except Exception as e:
verbose_proxy_logger.warning(
"ContentFilterGuardrail: failed to init competitor intent checker: %s",
e,
)
# Load categories if provided
if categories:
self._load_categories(categories)
@ -1158,9 +1178,7 @@ class ContentFilterGuardrail(CustomGuardrail):
pattern_name=pattern_name.upper()
)
text = self._mask_spans(text, spans, redaction_tag)
verbose_proxy_logger.info(
f"Masked all {pattern_name} matches in content"
)
verbose_proxy_logger.info(f"Masked all {pattern_name} matches in content")
return text
@ -1383,7 +1401,9 @@ class ContentFilterGuardrail(CustomGuardrail):
masked_entity_count: Dictionary to update with counts
"""
for detection in detections:
if detection["action"] == ContentFilterAction.MASK.value:
if detection.get("type") == "competitor_intent":
continue
if detection.get("action") == ContentFilterAction.MASK.value:
detection_type = detection["type"]
if detection_type == "pattern":
pattern_detection = cast(PatternDetection, detection)
@ -1409,18 +1429,27 @@ class ContentFilterGuardrail(CustomGuardrail):
"""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"]}
action_taken = detection.get("action", detection.get("action_hint", ""))
detail: dict = {"type": detection["type"], "action_taken": action_taken}
if detection["type"] == "pattern":
detail["detection_method"] = "regex"
detail["snippet"] = cast(PatternDetection, detection).get("pattern_name", "")
detail["snippet"] = cast(PatternDetection, detection).get(
"pattern_name", ""
)
elif detection["type"] == "blocked_word":
detail["detection_method"] = "keyword"
detail["snippet"] = cast(BlockedWordDetection, detection).get("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", "")
elif detection["type"] == "competitor_intent":
detail["detection_method"] = "intent"
detail["snippet"] = detection.get("intent", "")
detail["confidence"] = detection.get("confidence")
match_details.append(detail)
return match_details
@ -1430,19 +1459,28 @@ class ContentFilterGuardrail(CustomGuardrail):
for detection in detections:
if detection["type"] == "pattern":
methods.add("regex")
elif detection["type"] == "competitor_intent":
methods.add("intent")
else:
methods.add("keyword")
return ",".join(sorted(methods)) if methods else ""
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)
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()]
names = [
cat.description or cat.category_name
for cat in self.loaded_categories.values()
]
return ", ".join(names) if names else None
def _compute_risk_score(
@ -1480,6 +1518,72 @@ class ContentFilterGuardrail(CustomGuardrail):
return round(min(10.0, score), 1)
def _apply_competitor_intent_policy(
self,
intent_result: CompetitorIntentResult,
request_data: dict,
detections: List[ContentFilterDetection],
) -> None:
"""
Apply policy for competitor intent result: refuse (raise), reframe (passthrough), or log_only/allow (return).
Appends competitor_intent detection to detections. Never returns for refuse/reframe.
"""
intent_val = intent_result.get("intent", "other")
confidence_val = intent_result.get("confidence", 0.0)
action_hint_val = intent_result.get("action_hint", "allow")
evidence_list = intent_result.get("evidence", [])
detection: CompetitorIntentDetection = {
"type": "competitor_intent",
"intent": intent_val,
"confidence": confidence_val,
"action_hint": action_hint_val,
"entities": intent_result.get("entities", {}),
"signals": intent_result.get("signals", []),
"evidence": [dict(e) for e in evidence_list],
}
detections.append(detection)
if action_hint_val == "refuse":
msg = "Content blocked: competitor comparison or ranking intent detected."
if self._competitor_intent_checker and getattr(
self._competitor_intent_checker, "refuse_message_template", None
):
msg = self._competitor_intent_checker.refuse_message_template or msg
verbose_proxy_logger.warning(
"ContentFilterGuardrail: competitor intent refuse - %s", intent_val
)
raise HTTPException(
status_code=403,
detail={
"error": msg,
"intent": intent_val,
"confidence": confidence_val,
},
)
if action_hint_val == "reframe":
msg = (
"I can help with questions about our products and services. "
"Would you like to compare specific features or get more information?"
)
if self._competitor_intent_checker and getattr(
self._competitor_intent_checker, "reframe_message_template", None
):
msg = self._competitor_intent_checker.reframe_message_template or msg
verbose_proxy_logger.info(
"ContentFilterGuardrail: competitor intent reframe - %s", intent_val
)
self.raise_passthrough_exception(
violation_message=msg,
request_data=request_data,
detection_info=dict(intent_result),
)
# log_only or allow: just log (detection already appended)
verbose_proxy_logger.debug(
"ContentFilterGuardrail: competitor intent %s (action_hint=%s)",
intent_val,
action_hint_val,
)
def _log_guardrail_information(
self,
request_data: dict,
@ -1511,6 +1615,28 @@ class ContentFilterGuardrail(CustomGuardrail):
else [dict(detection) for detection in detections]
)
# Competitor intent: add confidence and classification to tracing if present
tracing_kw: Dict[str, Any] = {
"guardrail_id": self.config_guardrail_id or self.guardrail_name,
"policy_template": self.config_policy_template
or 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(),
"risk_score": self._compute_risk_score(
detections, masked_entity_count, status
),
}
for d in detections:
if isinstance(d, dict) and d.get("type") == "competitor_intent":
tracing_kw["confidence_score"] = d.get("confidence")
tracing_kw["classification"] = dict(d)
break
self.add_standard_logging_guardrail_information_to_request_data(
guardrail_provider=self.guardrail_provider,
guardrail_json_response=guardrail_json_response,
@ -1520,14 +1646,7 @@ class ContentFilterGuardrail(CustomGuardrail):
end_time=datetime.now().timestamp(),
duration=(datetime.now() - start_time).total_seconds(),
masked_entity_count=masked_entity_count,
tracing_detail=GuardrailTracingDetail(
guardrail_id=self.config_guardrail_id or self.guardrail_name,
policy_template=self.config_policy_template or 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(),
risk_score=self._compute_risk_score(detections, masked_entity_count, status),
),
tracing_detail=GuardrailTracingDetail(**tracing_kw),
)
async def apply_guardrail(
@ -1575,6 +1694,13 @@ class ContentFilterGuardrail(CustomGuardrail):
processed_texts = []
for text in texts:
# Competitor intent check first (optional; may refuse/reframe)
if self._competitor_intent_checker and text:
intent_result = self._competitor_intent_checker.run(text)
if intent_result.get("intent", "other") != "other":
self._apply_competitor_intent_policy(
intent_result, request_data, detections
)
filtered_text = self._filter_single_text(text, detections=detections)
processed_texts.append(filtered_text)
@ -1700,4 +1826,4 @@ class ContentFilterGuardrail(CustomGuardrail):
LitellmContentFilterGuardrailConfigModel,
)
return LitellmContentFilterGuardrailConfigModel
return LitellmContentFilterGuardrailConfigModel

View file

@ -0,0 +1,65 @@
# Content Filter Examples
## Industry-specific competitor intent: Emirates (airline)
The **generic** competitor intent blocker only needs:
- **brand_self**: Your brand names and aliases (e.g. `["emirates", "ek"]`)
- **competitors**: List of competitor names (e.g. `["qatar airways", "etihad"]`)
To make it effective for a specific industry (e.g. airlines), add an **industry layer** on top:
1. **domain_words** – Terms that signal “this is about our vertical.”
For Emirates (airline): `airline`, `carrier`, `flight`, `business class`, `lounge`, etc.
This enables the **category_ranking** path (e.g. “Which Gulf airline is the best?”) and the scoring **gate** (so “best” alone doesn’t trigger without domain/geo).
2. **route_geo_cues** – Optional geography/hub terms.
For Emirates: `doha`, `dubai`, `abu dhabi`, `gulf`, `middle east`.
3. **descriptor_lexicon** – Phrases that count as indirect competitor reference.
For aviation: `doha airline`, `oryx airline`, `gulf carrier`, `five star airline`, `skytrax`.
4. **competitor_aliases** – Per-competitor aliases (IATA codes, nicknames).
Example: `qatar airways` → `["qr", "doha airline"]`, `etihad` → `["ey"]`.
5. **policy** – What to do per intent band: `refuse`, `reframe`, `log_only`, or `allow`.
Example: `competitor_comparison: refuse`, `category_ranking: reframe`.
See **emirates_competitor_intent_guardrail.yaml** for a full example you can copy into your proxy `guardrails` config or merge into an existing `litellm_content_filter` guardrail.
### Using the example in your proxy config
In `config.yaml`:
```yaml
guardrails:
- guardrail_name: "emirates-competitor-intent"
litellm_params:
guardrail: litellm_content_filter
mode: pre_call
competitor_intent_config:
brand_self: [emirates, ek]
competitors: [qatar airways, qatar, etihad, turkish airlines]
domain_words: [airline, carrier, flight, business class, lounge]
route_geo_cues: [doha, dubai, abu dhabi, gulf]
descriptor_lexicon: [doha airline, gulf carrier, five star airline]
competitor_aliases:
qatar airways: [qr, doha airline]
etihad: [ey]
policy:
competitor_comparison: refuse
possible_competitor_comparison: reframe
category_ranking: reframe
log_only: log_only
```
Then attach this guardrail to your router/policy (e.g. `guardrails.add: [emirates-competitor-intent]`).
### Other industries
Use the same pattern:
- **SaaS**: `domain_words`: `["platform", "tool", "solution", "integration"]`; optional `route_geo_cues` if regional.
- **Retail**: `domain_words`: `["store", "brand", "product line"]`; `competitor_aliases` for brand nicknames.
The implementation is generic; only the config (and optional industry presets) are industry-specific.

View file

@ -0,0 +1,134 @@
# Emirates (airline) industry-specific competitor intent guardrail
#
# This is an example of building on the generic competitor intent blocker with
# industry-specific inputs: domain_words, route_geo_cues, descriptor_lexicon,
# and competitor_aliases for aviation.
#
# Use this guardrail in your proxy config under guardrails.litellm_params, or
# merge competitor_intent_config into an existing litellm_content_filter guardrail.
#
# Minimal required (generic): brand_self, competitors
# Industry layer (airline): domain_words, route_geo_cues, descriptor_lexicon,
# competitor_aliases (IATA/descriptive), optional indirect_competitor_patterns
guardrail_name: "emirates-competitor-intent"
guardrail_id: "emirates-competitor-intent"
litellm_params:
guardrail: litellm_content_filter
mode: pre_call
default_on: false
# Generic inputs (customer provides)
competitor_intent_config:
# Your brand (Emirates) and how users refer to it
brand_self:
- emirates
- ek
# Direct competitors (canonical names; use competitor_aliases for codes/typos)
competitors:
- qatar airways
- qatar
- etihad
- turkish airlines
- singapore airlines
- british airways
# Geo entities: countries, cities, airports (for other-meaning vs airline disambiguation)
locations:
- qatar
- doha
- doh
- hamad international
- lusail
# Industry: airline domain words (trigger category_ranking / gate)
domain_words:
- airline
- airlines
- carrier
- flight
- flights
- business class
- first class
- lounge
- lounges
- premium cabin
- long haul
# Industry: route/geo cues (Gulf, hubs)
route_geo_cues:
- doha
- dubai
- abu dhabi
- gulf
- middle east
# Indirect references (descriptor lexicon)
descriptor_lexicon:
- doha airline
- oryx airline
- gulf carrier
- five star airline
- skytrax
# Per-competitor aliases (IATA codes, nicknames)
competitor_aliases:
qatar airways:
- qr
- doha airline
qatar:
- qr
etihad:
- ey
- abu dhabi airline
emirates:
- ek
- dubai airline
# Optional: regex patterns for indirect competitor mention (on normalized text)
# indirect_competitor_patterns:
# qatar airways:
# - "the doha airline"
# - "doha carrier"
# etihad:
# - "abu dhabi carrier"
# Policy: what to do per intent band
policy:
competitor_comparison: refuse
possible_competitor_comparison: reframe
category_ranking: reframe
log_only: log_only
threshold_high: 0.70
threshold_medium: 0.45
threshold_low: 0.30
# Optional: custom messages
reframe_message_template: >
I'm here to help with Emirates services, bookings, and travel information.
Would you like to know about our routes, cabin classes, or loyalty program?
refuse_message_template: "We can't compare Emirates with other airlines. How can I help with Emirates travel?"
# --- How to use in config.yaml ---
#
# guardrails:
# - guardrail_name: "emirates-competitor-intent"
# litellm_params:
# guardrail: litellm_content_filter
# mode: pre_call
# competitor_intent_config:
# brand_self: [emirates, ek]
# competitors: [qatar airways, etihad, ...]
# domain_words: [airline, carrier, flight, ...]
# route_geo_cues: [doha, dubai, abu dhabi]
# descriptor_lexicon: [doha airline, gulf carrier, ...]
# competitor_aliases:
# qatar airways: [qr, doha airline]
# etihad: [ey]
# policy:
# competitor_comparison: refuse
# possible_competitor_comparison: reframe
# category_ranking: reframe
# log_only: log_only

View file

@ -1,10 +1,42 @@
from enum import Enum
from typing import List, Literal, Optional, TypedDict, Union
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
from pydantic import Field
from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
from litellm.types.proxy.guardrails.guardrail_hooks.base import \
GuardrailConfigModel
# --- Competitor intent blocker (generic, industry-agnostic) ---
CompetitorIntentType = Literal[
"competitor_comparison",
"possible_competitor_comparison",
"category_ranking",
"log_only",
"other",
]
CompetitorActionHint = Literal["allow", "reframe", "refuse", "escalate", "log_only"]
class CompetitorIntentEvidenceEntry(TypedDict, total=False):
"""Single evidence entry: what matched and what it resolved to."""
type: Literal["entity", "signal"]
key: str # e.g. "competitor", "ranking", "brand_self"
value: Optional[str] # resolved canonical value (e.g. "qatar_airways")
match: str # matched substring
class CompetitorIntentResult(TypedDict, total=False):
"""Structured output from competitor intent checker."""
intent: CompetitorIntentType
confidence: float
entities: Dict[str, List[str]] # brand_self, competitors, category
signals: List[str]
action_hint: CompetitorActionHint
evidence: List[CompetitorIntentEvidenceEntry]
# Detection type enum
@ -37,7 +69,24 @@ class CategoryKeywordDetection(TypedDict):
action: str # ContentFilterAction.value
ContentFilterDetection = Union[PatternDetection, BlockedWordDetection, CategoryKeywordDetection]
class CompetitorIntentDetection(TypedDict):
"""Detection from competitor intent checker (intent + evidence)."""
type: Literal["competitor_intent"]
intent: str
confidence: float
action_hint: str
entities: Dict[str, List[str]]
signals: List[str]
evidence: List[Dict[str, Any]]
ContentFilterDetection = Union[
PatternDetection,
BlockedWordDetection,
CategoryKeywordDetection,
CompetitorIntentDetection,
]
class ContentFilterCategoryConfig(BaseLiteLLMOpenAIResponseObject):
@ -113,6 +162,17 @@ class LitellmContentFilterGuardrailConfigModel(GuardrailConfigModel):
description="Tag to use for keyword redaction",
)
# Competitor intent blocker (generic; industry presets add domain_words, etc.)
competitor_intent_config: Optional[Dict[str, Any]] = Field(
default=None,
description="Optional config for intent-based competitor comparison detection. "
"Keys: brand_self (list), competitors (list), competitor_aliases (dict), "
"domain_words (list, optional), route_geo_cues (list, optional), "
"descriptor_lexicon (list, optional), indirect_competitor_patterns (dict, optional), "
"policy (dict), threshold_high, threshold_medium, threshold_low, "
"reframe_message_template, refuse_message_template.",
)
@staticmethod
def ui_friendly_name() -> str:
return "LiteLLM Content Filter"

View file

@ -0,0 +1,165 @@
"""
Tests for competitor intent detection (normalize, entity layer, scoring, policy).
"""
import pytest
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.competitor_intent import (
AirlineCompetitorIntentChecker,
normalize,
text_for_entity_matching,
)
class TestNormalize:
"""Test text normalization (leetspeak, spacing, zero-width)."""
def test_normalize_lowercase(self):
assert normalize("Is Qatar Better?") == "is qatar better?"
def test_normalize_leetspeak(self):
assert "qatar" in normalize("q@tar")
assert "qatar" in normalize("q4tar")
def test_normalize_collapse_whitespace(self):
assert normalize("hello world") == "hello world"
def test_normalize_spaced_out_letters(self):
# Single-letter tokens collapsed into word
assert "qatar" in normalize("q a t a r").replace(" ", "")
def test_normalize_empty(self):
assert normalize("") == ""
assert normalize(None) == ""
def test_text_for_entity_matching_removes_punctuation(self):
t = text_for_entity_matching("q.a.t.a.r emirates")
assert "emirates" in t
assert "." not in t
class TestAirlineCompetitorIntentChecker:
"""Test AirlineCompetitorIntentChecker run() and intent bands."""
@pytest.fixture
def generic_config(self):
return {
"brand_self": ["emirates", "ek"],
"competitors": ["qatar airways", "etihad", "qatar"],
"competitor_aliases": {
"qatar airways": ["qr", "doha airline"],
"qatar": ["qr"],
},
"locations": ["qatar", "doha", "doh"],
"domain_words": ["airline", "carrier", "flight", "business class"],
"route_geo_cues": ["doha", "dubai", "abu dhabi"],
"policy": {
"competitor_comparison": "refuse",
"possible_competitor_comparison": "reframe",
"category_ranking": "reframe",
"log_only": "log_only",
},
"threshold_high": 0.70,
"threshold_medium": 0.45,
"threshold_low": 0.30,
}
def test_run_other_intent(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("What is the weather today?")
assert result["intent"] == "other"
assert result["action_hint"] == "allow"
def test_run_competitor_comparison_direct(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("Is Qatar better than Emirates?")
assert result["intent"] in ("competitor_comparison", "possible_competitor_comparison")
assert "competitor_entity" in result.get("signals", []) or "competitors" in str(result.get("entities", {}))
assert result["confidence"] >= 0.45
def test_run_competitor_comparison_as_good_as(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("Is Qatar as good as Emirates?")
assert result["intent"] != "other"
assert result["confidence"] >= 0.45
def test_run_ranking_with_competitor(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("Why is Qatar Airways the best?")
assert result["intent"] != "other"
assert "qatar" in str(result.get("entities", {}).get("competitors", [])).lower() or "competitor" in str(result.get("signals", []))
def test_run_ranking_without_competitor_category_ranking(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("Which Gulf airline is the best?")
# domain_words "airline" + ranking "best" + geo "gulf" not in route_geo_cues but "airline" is domain
assert result["intent"] in ("category_ranking", "possible_competitor_comparison", "log_only", "other")
def test_run_evidence_populated(self, generic_config):
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("Is Qatar better than Emirates?")
assert "evidence" in result
assert isinstance(result["evidence"], list)
def test_run_gate_prevents_false_positive(self, generic_config):
# "best" alone without entity or domain should not trigger competitor_comparison
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("What is the best way to cook pasta?")
assert result["intent"] in ("other", "log_only")
def test_other_meaning_context_suppression(self, generic_config):
# "flights to qatar" = other meaning (country), not competitor airline
checker = AirlineCompetitorIntentChecker(generic_config)
result = checker.run("how expensive are flights to qatar?")
assert result["intent"] == "other"
assert not result.get("entities", {}).get("competitors")
class TestContentFilterWithCompetitorIntent:
"""Integration: ContentFilterGuardrail with competitor_intent_config."""
@pytest.mark.asyncio
async def test_apply_guardrail_with_competitor_intent_allow(self):
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
ContentFilterGuardrail,
)
guardrail = ContentFilterGuardrail(
guardrail_name="test-competitor",
competitor_intent_config={
"brand_self": ["emirates"],
"competitors": ["qatar"],
"domain_words": ["airline"],
"policy": {"competitor_comparison": "refuse", "possible_competitor_comparison": "reframe"},
},
)
inputs = {"texts": ["What is the capital of France?"]}
result = await guardrail.apply_guardrail(
inputs, request_data={}, input_type="request"
)
assert result["texts"] == ["What is the capital of France?"]
@pytest.mark.asyncio
async def test_apply_guardrail_with_competitor_intent_refuse(self):
from fastapi import HTTPException
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
ContentFilterGuardrail,
)
guardrail = ContentFilterGuardrail(
guardrail_name="test-competitor",
competitor_intent_config={
"brand_self": ["emirates"],
"competitors": ["qatar airways"],
"domain_words": ["airline", "flight"],
"policy": {"competitor_comparison": "refuse"},
"threshold_high": 0.5,
},
)
inputs = {"texts": ["Is Qatar Airways better than Emirates?"]}
with pytest.raises(HTTPException) as exc_info:
await guardrail.apply_guardrail(
inputs, request_data={}, input_type="request"
)
assert exc_info.value.status_code == 403

View file

@ -250,10 +250,254 @@ const compliancePrompts: CompliancePrompt[] = [
),
];
export const airlineCompliancePrompts: CompliancePrompt[] = [
// =====================
// DESTINATION (should PASS)
// =====================
{
id: "airline_001",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "User is asking about a travel destination, not a competitor airline.",
prompt: "Do you have flights to Qatar?",
expectedResult: "pass",
},
{
id: "airline_002",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Location-based travel planning.",
prompt: "How long is the flight to Doha?",
expectedResult: "pass",
},
{
id: "airline_003",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Transit and visa questions for a country.",
prompt: "Do I need a visa to transit through Qatar?",
expectedResult: "pass",
},
{
id: "airline_004",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Airport facilities question.",
prompt: "Are there lounges at Hamad Airport?",
expectedResult: "pass",
},
{
id: "airline_005",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Travel logistics to a location.",
prompt: "What documents are required to enter Singapore?",
expectedResult: "pass",
},
{
id: "airline_006",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Flight planning for a country destination.",
prompt: "What airlines fly to Turkey?",
expectedResult: "pass",
},
{
id: "airline_007",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Airport code information.",
prompt: "What is the airport code for Doha?",
expectedResult: "pass",
},
{
id: "airline_008",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "General travel planning question.",
prompt: "Is Doha a good place for a long layover?",
expectedResult: "pass",
},
{
id: "airline_009",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Country weather question.",
prompt: "What is the weather like in Qatar in March?",
expectedResult: "pass",
},
{
id: "airline_010",
framework: "Airline Brand Protection",
category: "Destination Intent",
categoryIcon: "✈️",
categoryDescription: "Transit timing question.",
prompt: "How much connection time do I need in Doha?",
expectedResult: "pass",
},
// =====================
// COMPETITOR (should FAIL)
// =====================
{
id: "airline_011",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Direct comparison between airlines.",
prompt: "Is Qatar Airways better than Emirates?",
expectedResult: "fail",
},
{
id: "airline_012",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Comparison request for premium cabins.",
prompt: "Which airline has better business class, Qatar or Singapore Airlines?",
expectedResult: "fail",
},
{
id: "airline_013",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Airline ranking request.",
prompt: "Which airline is the best for long haul flights?",
expectedResult: "fail",
},
{
id: "airline_014",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Decision-making between competitors.",
prompt: "Should I choose Qatar Airways or another airline?",
expectedResult: "fail",
},
{
id: "airline_015",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "General airline comparison.",
prompt: "Compare Turkish Airlines vs Lufthansa.",
expectedResult: "fail",
},
{
id: "airline_016",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Service quality comparison.",
prompt: "Which airline has better lounges?",
expectedResult: "fail",
},
{
id: "airline_017",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Customer review comparison.",
prompt: "Which airline has the best customer satisfaction?",
expectedResult: "fail",
},
{
id: "airline_018",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Indirect competitor reference via location.",
prompt: "Is the Doha airline better than other carriers?",
expectedResult: "fail",
},
{
id: "airline_019",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Switching intent to competitor.",
prompt: "Should I switch to another airline for a better experience?",
expectedResult: "fail",
},
{
id: "airline_020",
framework: "Airline Brand Protection",
category: "Competitor Comparison",
categoryIcon: "⚠️",
categoryDescription: "Global airline ranking request.",
prompt: "Which airline is ranked number one worldwide?",
expectedResult: "fail",
},
// =====================
// AMBIGUOUS (should PASS but needs careful handling)
// =====================
{
id: "airline_021",
framework: "Airline Brand Protection",
category: "Ambiguous Entity",
categoryIcon: "❓",
categoryDescription: "Could refer to airline or country.",
prompt: "Qatar baggage allowance",
expectedResult: "pass",
},
{
id: "airline_022",
framework: "Airline Brand Protection",
category: "Ambiguous Entity",
categoryIcon: "❓",
categoryDescription: "Ambiguous lounge reference.",
prompt: "Qatar lounge access rules",
expectedResult: "pass",
},
{
id: "airline_023",
framework: "Airline Brand Protection",
category: "Ambiguous Entity",
categoryIcon: "❓",
categoryDescription: "Ambiguous operational query.",
prompt: "Qatar check in time",
expectedResult: "pass",
},
{
id: "airline_024",
framework: "Airline Brand Protection",
category: "Ambiguous Entity",
categoryIcon: "❓",
categoryDescription: "Ambiguous airline vs destination reference.",
prompt: "Doha premium lounge access",
expectedResult: "pass",
},
{
id: "airline_025",
framework: "Airline Brand Protection",
category: "Ambiguous Entity",
categoryIcon: "❓",
categoryDescription: "Ambiguous refund context.",
prompt: "Qatar refund policy",
expectedResult: "pass",
},
];
const allCompliancePrompts: CompliancePrompt[] = [
...compliancePrompts,
...airlineCompliancePrompts,
];
export function getCategories(): ComplianceCategory[] {
const categoryMap = new Map<string, ComplianceCategory>();
for (const prompt of compliancePrompts) {
for (const prompt of allCompliancePrompts) {
if (!categoryMap.has(prompt.category)) {
categoryMap.set(prompt.category, {
name: prompt.category,
@ -277,6 +521,10 @@ const frameworkMeta: Record<string, { icon: string; description: string }> = {
icon: "lock",
description: "General Data Protection Regulation — data privacy and protection requirements.",
},
"Airline Brand Protection": {
icon: "plane",
description: "Destination vs competitor intent — avoid answering competitor comparison questions.",
},
};
export function getFrameworks(): ComplianceFramework[] {
@ -285,7 +533,7 @@ export function getFrameworks(): ComplianceFramework[] {
{ categories: Map<string, ComplianceCategory> }
>();
for (const prompt of compliancePrompts) {
for (const prompt of allCompliancePrompts) {
if (!frameworkMap.has(prompt.framework)) {
frameworkMap.set(prompt.framework, { categories: new Map() });
}

View file

@ -14,7 +14,7 @@
"moduleResolution": "bundler",
"resolveJsonModule": true,
"isolatedModules": true,
"jsx": "react-jsx",
"jsx": "preserve",
"incremental": true,
"plugins": [
{