Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/competitor_intent/base.py: 11%
123 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Generic competitor intent checker: two entity sets and overridable disambiguation.
3"""
5import re
6import unicodedata
7from re import Pattern
8from typing import Any, Final, cast
10from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
11 CompetitorActionHint,
12 CompetitorIntentEvidenceEntry,
13 CompetitorIntentResult,
14 CompetitorIntentType,
15)
17ZERO_WIDTH: Final = re.compile(r"[\u200b-\u200d\u2060\ufeff]")
18LEET: Final = {"@": "a", "4": "a", "0": "o", "3": "e", "1": "i", "5": "s", "7": "t"}
20OTHER_MEANING_DEFAULT_THRESHOLD: Final = 0.65 # Below this → treat as non-competitor (safe default).
23def normalize(text: str) -> str:
24 """Lowercase, NFKC, strip zero-width, leetspeak, collapse spaces."""
25 if not text or not isinstance(text, str):
26 return ""
27 t = ZERO_WIDTH.sub("", text)
28 t = unicodedata.normalize("NFKC", t).lower().strip()
29 for c, r in LEET.items():
30 t = t.replace(c, r)
31 return re.sub(r"\s+", " ", t)
34def _word_boundary_match(text: str, token: str) -> bool:
35 """True if token appears as a word in text."""
36 return bool(re.search(r"\b" + re.escape(token) + r"\b", text))
39def _count_signals(text: str, patterns: list[str]) -> int:
40 """Count how many of the patterns appear in text."""
41 return sum(1 for p in patterns if re.search(p, text, re.IGNORECASE))
44def _compile_marker(pattern: str | None) -> Pattern[str] | None:
45 """Compile optional regex string to a pattern."""
46 if not pattern or not pattern.strip():
47 return None
48 try:
49 return re.compile(pattern, re.IGNORECASE)
50 except re.error:
51 return None
54def text_for_entity_matching(text: str) -> str:
55 """Letters-only variant for entity matching (e.g. split punctuation)."""
56 t: Final = re.sub(r"[^\w\s]", " ", text)
57 return re.sub(r"\s+", " ", t).strip()
60class BaseCompetitorIntentChecker:
61 """
62 Generic competitor intent checker with two entity sets. Ambiguous tokens
63 (competitor + other-meaning, e.g. location) are classified by overridable
64 _classify_ambiguous(). Base implementation: treat as non-competitor.
65 """
67 def __init__(self, config: dict[str, Any]) -> None:
68 self.brand_self: list[str] = [s.lower().strip() for s in (config.get("brand_self") or []) if s]
69 competitors: Final[list[str]] = [s.lower().strip() for s in (config.get("competitors") or []) if s]
70 aliases_map: Final[dict[str, list[str]]] = config.get("competitor_aliases") or {}
71 self.competitor_canonical: dict[str, str] = {}
72 self._competitor_tokens: set[str] = set()
73 for c in competitors:
74 self._competitor_tokens.add(c)
75 self.competitor_canonical[c] = c
76 for a in aliases_map.get(c) or []:
77 a = a.lower().strip()
78 if a:
79 self._competitor_tokens.add(a)
80 self.competitor_canonical[a] = c
82 other: Final[list[str]] = [s.lower().strip() for s in (config.get("locations") or []) if s]
83 self._other_meaning_tokens: set[str] = set(other)
84 self._ambiguous: set[str] = self._competitor_tokens & self._other_meaning_tokens
86 self.policy: dict[str, str] = config.get("policy") or {}
87 self.threshold_high = float(config.get("threshold_high", 0.70))
88 self.threshold_medium = float(config.get("threshold_medium", 0.45))
89 self.threshold_low = float(config.get("threshold_low", 0.30))
90 self.reframe_message_template: str | None = config.get("reframe_message_template")
91 self.refuse_message_template: str | None = config.get("refuse_message_template")
92 self._comparison_words: list[str] = list(
93 config.get("comparison_words")
94 or [
95 "better",
96 "worse",
97 "best",
98 "vs",
99 "versus",
100 "compare",
101 "alternative",
102 "recommend",
103 "ranked",
104 ]
105 )
106 self._domain_words: list[str] = [s.lower().strip() for s in (config.get("domain_words") or []) if s]
108 def _classify_ambiguous(self, text: str, token: str) -> tuple[str, float]:
109 """
110 Override in subclasses for industry-specific logic. Base: treat as non-competitor.
111 """
112 return "OTHER_MEANING", 0.5
114 def _find_matches(self, text: str) -> list[tuple[str, str, bool]]:
115 """Find competitor matches; mark ambiguous (also in other-meaning set)."""
116 normalized: Final = normalize(text)
117 found: Final[list[tuple[str, str, bool]]] = []
118 seen: Final[set[tuple[str, str]]] = set()
119 for token in self._competitor_tokens:
120 if not _word_boundary_match(normalized, token):
121 continue
122 canonical = self.competitor_canonical.get(token, token)
123 key = (token, canonical)
124 if key in seen:
125 continue
126 seen.add(key)
127 is_ambig = token in self._ambiguous or token in self._other_meaning_tokens
128 found.append((token, canonical, is_ambig))
129 return found
131 def run(self, text: str) -> CompetitorIntentResult:
132 """Classify competitor intent; non-competitor when ambiguous or low confidence."""
133 normalized: Final = normalize(text)
134 evidence: Final[list[CompetitorIntentEvidenceEntry]] = []
135 entities: Final[dict[str, list[str]]] = {
136 "brand_self": [],
137 "competitors": [],
138 "category": [],
139 }
141 for b in self.brand_self:
142 if _word_boundary_match(normalized, b):
143 entities["brand_self"].append(b)
144 evidence.append({"type": "entity", "key": "brand_self", "value": b, "match": b})
146 matches: Final = self._find_matches(text)
147 if not matches:
148 has_comparison = any(re.search(r"\b" + re.escape(w) + r"\b", normalized) for w in self._comparison_words)
149 has_domain: Final = self._domain_words and any(
150 re.search(r"\b" + re.escape(w) + r"\b", normalized) for w in self._domain_words
151 )
152 if has_comparison and has_domain:
153 evidence.append(
154 {
155 "type": "signal",
156 "key": "category_ranking",
157 "match": "comparison + domain",
158 }
159 )
160 action_hint: Final = cast(
161 CompetitorActionHint,
162 self.policy.get("category_ranking", "reframe"),
163 )
164 return {
165 "intent": "category_ranking",
166 "confidence": 0.65,
167 "entities": entities,
168 "signals": ["category_ranking"],
169 "action_hint": action_hint,
170 "evidence": evidence,
171 }
172 return {
173 "intent": "other",
174 "confidence": 0.0,
175 "entities": entities,
176 "signals": [],
177 "action_hint": "allow",
178 "evidence": evidence,
179 }
181 competitor_resolved: Final[list[str]] = []
182 for token, canonical, _ in matches:
183 label, conf = self._classify_ambiguous(normalized, token)
184 if label == "OTHER_MEANING":
185 evidence.append({"type": "signal", "key": "other_meaning", "match": token})
186 continue
187 if label == "COMPETITOR":
188 competitor_resolved.append(canonical)
189 evidence.append(
190 {
191 "type": "entity",
192 "key": "competitor",
193 "value": canonical,
194 "match": token,
195 }
196 )
197 if conf < OTHER_MEANING_DEFAULT_THRESHOLD:
198 competitor_resolved.pop()
199 evidence.append(
200 {
201 "type": "signal",
202 "key": "other_meaning_default",
203 "match": f"confidence {conf:.2f}",
204 }
205 )
206 continue
208 entities["competitors"] = list(dict.fromkeys(competitor_resolved))
210 if not competitor_resolved:
211 return {
212 "intent": "other",
213 "confidence": 0.0,
214 "entities": entities,
215 "signals": ["other_meaning_or_ambiguous"],
216 "action_hint": "allow",
217 "evidence": evidence,
218 }
220 has_comparison = any(re.search(r"\b" + re.escape(w) + r"\b", normalized) for w in self._comparison_words)
221 if has_comparison:
222 evidence.append({"type": "signal", "key": "comparison", "match": "comparison language"})
223 confidence: Final = 0.75 if has_comparison else 0.55
224 if confidence >= self.threshold_high:
225 intent = "competitor_comparison"
226 elif confidence >= self.threshold_medium:
227 intent = "possible_competitor_comparison"
228 elif confidence >= self.threshold_low:
229 intent = "log_only"
230 else:
231 intent = "other"
233 resolved_action_hint: CompetitorActionHint = cast(CompetitorActionHint, self.policy.get(intent, "allow"))
234 if intent == "log_only":
235 resolved_action_hint = "log_only"
236 if intent == "other":
237 resolved_action_hint = "allow"
239 return {
240 "intent": cast(CompetitorIntentType, intent),
241 "confidence": round(confidence, 2),
242 "entities": entities,
243 "signals": ["competitor_resolved"] + (["comparison"] if has_comparison else []),
244 "action_hint": resolved_action_hint,
245 "evidence": evidence,
246 }