Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/competitor_intent/airline.py: 13%

80 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 12:01 +0000

1""" 

2Airline-specific competitor intent: other meaning (e.g. location/travel context) vs competitor airline. 

3 

4Uses context-based disambiguation only: no hardcoded place lists. Detects travel-location 

5language (prepositions, travel verbs, booking/entry nouns) vs airline context (airways, 

6carrier, lounge, miles, etc.) and scores to decide OTHER_MEANING vs COMPETITOR. 

7 

8When competitors is not provided, loads major_airlines.json and excludes the customer's 

9brand_self so all other major airlines are treated as competitors. 

10""" 

11 

12import json 

13from pathlib import Path 

14from typing import Any, Final 

15 

16from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.competitor_intent.base import ( 

17 BaseCompetitorIntentChecker, 

18 _compile_marker, 

19 _count_signals, 

20 _word_boundary_match, 

21) 

22 

23# Location/travel context: prepositions, travel verbs, booking nouns, entry/geo nouns. 

24# No place-name list; these patterns detect "destination context" generically. 

25AIRLINE_OTHER_MEANING_SIGNALS: Final = [ 

26 # Travel verb + preposition (e.g. "fly to", "layover in") 

27 r"\b(fly|flying|travel|traveling|going|visit|visiting|transit|layover|stopover)\b.{0,12}\b(to|from|via|in|at|through|into)\b", 

28 # Booking + preposition 

29 r"\bflight(s)?\b.{0,10}\b(to|from|via)\b", 

30 r"\bticket(s)?\b.{0,8}\b(to|for)\b", 

31 r"\bfare(s)?\b.{0,8}\b(to)\b", 

32 # Entry/geo/booking single words 

33 r"\bvisa\b", 

34 r"\bimmigration\b", 

35 r"\bcustoms\b", 

36 r"\bentry\b", 

37 r"\bairport\b", 

38 r"\bterminal\b", 

39 r"\bgate\b", 

40 r"\bdeparture\b", 

41 r"\barrival\b", 

42 r"\bitinerary\b", 

43 r"\bweather\b", 

44 r"\bhotel\b", 

45 r"\bcity\b", 

46 # Prepositions alone (weaker; often near a place) 

47 r"\bto\s+", 

48 r"\bfrom\s+", 

49 r"\bin\s+", 

50 r"\bat\s+", 

51 r"\bvia\s+", 

52] 

53 

54# Airline context: carrier/airline language, cabin, loyalty, operations. 

55# If ambiguous token appears near these → treat as COMPETITOR. 

56AIRLINE_COMPETITOR_SIGNALS: Final = [ 

57 r"\bairways?\b", 

58 r"\bairline\b", 

59 r"\bcarrier\b", 

60 r"\bcabin\s+crew\b", 

61 r"\bflight\s+attendant\b", 

62 r"\bbusiness\s+class\b", 

63 r"\bfirst\s+class\b", 

64 r"\beconomy\b", 

65 r"\blounge\b", 

66 r"\bbaggage\s+allowance\b", 

67 r"\bcheck[- ]?in\b", 

68 r"\bmiles\b", 

69 r"\bloyalty\b", 

70 r"\bstatus\b", 

71 r"\bfrequent\s+flyer\b", 

72 r"\bfleet\b", 

73 r"\baircraft\b", 

74 # Comparison/ranking 

75 r"\bbetter\b", 

76 r"\bbest\b", 

77 r"\bgood\b", 

78 r"\bas\s+good\s+as\b", 

79 r"\bvs\.?\b", 

80 r"\bversus\b", 

81 r"\bcompare\b", 

82 r"\balternative\b", 

83 r"\bcompetitor\b", 

84 # Brand-specific (optional; config can extend) 

85 r"\bqmiles\b", 

86 r"\bprivilege\s+club\b", 

87] 

88 

89# Operational-only: baggage, lounge, check-in, refund (no comparison language). 

90# When only these appear with ambiguous token → treat as product query (OTHER_MEANING). 

91AIRLINE_OPERATIONAL_SIGNALS: Final = [ 

92 r"\bbaggage\s+allowance\b", 

93 r"\blounge\b", 

94 r"\bcheck[- ]?in\b", 

95 r"\brefund\b", 

96 r"\bpremium\s+lounge\b", 

97] 

98# Comparison language: if present with competitor signals → COMPETITOR. 

99AIRLINE_COMPARISON_SIGNALS: Final = [ 

100 r"\bbetter\b", 

101 r"\bbest\b", 

102 r"\bvs\.?\b", 

103 r"\bversus\b", 

104 r"\bcompare\b", 

105] 

106 

107# Explicit markers: strong override when present. 

108AIRLINE_EXPLICIT_COMPETITOR_MARKER: Final = r"\b(airways?|airline|carrier)\b" 

109AIRLINE_EXPLICIT_OTHER_MEANING_MARKER: Final = ( 

110 r"\b(fly|travel|going|visit|layover|stopover|transit)\b.{0,12}\b(to|in|via|from)\b.{0,8}\b" 

111) 

112 

113_MAJOR_AIRLINES_PATH: Final = Path(__file__).resolve().parent / "major_airlines.json" 

114 

115 

116def _load_competitors_excluding_brand(brand_self: list[str]) -> list[str]: 

117 """ 

118 Load competitor tokens from major_airlines.json (harm_toxic_abuse-style format). 

119 Exclude any airline whose id or match variants overlap with brand_self. 

120 Returns a flat list of match variants (pipe-separated values) from non-excluded airlines. 

121 """ 

122 brand_set: Final = {b.lower().strip() for b in brand_self if b} 

123 if not _MAJOR_AIRLINES_PATH.exists(): 

124 return [] 

125 try: 

126 with open(_MAJOR_AIRLINES_PATH, encoding="utf-8") as f: 

127 airlines: Final = json.load(f) 

128 except (json.JSONDecodeError, OSError): 

129 return [] 

130 result: Final[list[str]] = [] 

131 for entry in airlines: 

132 if not isinstance(entry, dict): 

133 continue 

134 match_str = entry.get("match") or "" 

135 variants = [v.strip().lower() for v in match_str.split("|") if v.strip()] 

136 words_in_match: set[str] = set() 

137 for v in variants: 

138 words_in_match.update(v.split()) 

139 if brand_set & words_in_match or any(v in brand_set for v in variants): 

140 continue 

141 result.extend(variants) 

142 return result 

143 

144 

145class AirlineCompetitorIntentChecker(BaseCompetitorIntentChecker): 

146 """ 

147 Disambiguates other meaning (e.g. country/city/airport) vs competitor airline 

148 (e.g. "Qatar" → country vs Qatar Airways). Overrides _classify_ambiguous 

149 with other_meaning/competitor signals and explicit markers. 

150 """ 

151 

152 def __init__(self, config: dict[str, Any]) -> None: 

153 merged: Final[dict[str, Any]] = dict(config) 

154 if not merged.get("other_meaning_signals"): 

155 merged["other_meaning_signals"] = AIRLINE_OTHER_MEANING_SIGNALS 

156 if not merged.get("competitor_signals"): 

157 merged["competitor_signals"] = AIRLINE_COMPETITOR_SIGNALS 

158 # Optional: no default place list; config can add other_meaning_anchors for extra patterns 

159 if "other_meaning_anchors" not in merged: 

160 merged["other_meaning_anchors"] = [] 

161 if not merged.get("explicit_competitor_marker"): 

162 merged["explicit_competitor_marker"] = AIRLINE_EXPLICIT_COMPETITOR_MARKER 

163 if not merged.get("explicit_other_meaning_marker"): 

164 merged["explicit_other_meaning_marker"] = AIRLINE_EXPLICIT_OTHER_MEANING_MARKER 

165 if not merged.get("domain_words"): 

166 merged["domain_words"] = ["airline", "airlines", "carrier"] 

167 if not merged.get("competitors"): 

168 merged["competitors"] = _load_competitors_excluding_brand(merged.get("brand_self") or []) 

169 super().__init__(merged) 

170 self._other_meaning_signals = list(merged.get("other_meaning_signals") or []) 

171 self._competitor_signals = list(merged.get("competitor_signals") or []) 

172 self._other_meaning_anchors = list(merged.get("other_meaning_anchors") or []) 

173 self._explicit_competitor_marker = _compile_marker(merged.get("explicit_competitor_marker")) 

174 self._explicit_other_meaning_marker = _compile_marker(merged.get("explicit_other_meaning_marker")) 

175 

176 def _classify_ambiguous(self, text: str, token: str) -> tuple[str, float]: 

177 """Other meaning vs competitor using airline signals and explicit markers.""" 

178 text_lower: Final = text.lower() 

179 if ( 

180 self._explicit_competitor_marker 

181 and self._explicit_competitor_marker.search(text_lower) 

182 and _word_boundary_match(text_lower, token.lower()) 

183 ): 

184 return "COMPETITOR", 0.85 

185 if self._explicit_other_meaning_marker and self._explicit_other_meaning_marker.search(text_lower): 

186 return "OTHER_MEANING", 0.85 

187 # Operational-only: baggage/lounge/check-in/refund with no comparison → product query 

188 has_comparison: Final = _count_signals(text_lower, AIRLINE_COMPARISON_SIGNALS) > 0 

189 operational_count: Final = _count_signals(text_lower, AIRLINE_OPERATIONAL_SIGNALS) 

190 if not has_comparison and operational_count > 0: 

191 return "OTHER_MEANING", 0.85 

192 # Score: location/travel context vs airline context (no place-name list) 

193 other_count = _count_signals(text_lower, self._other_meaning_signals) 

194 if self._other_meaning_anchors: 

195 other_count += _count_signals(text_lower, self._other_meaning_anchors) 

196 comp_count: Final = _count_signals(text_lower, self._competitor_signals) 

197 total: Final = other_count + comp_count 

198 if total == 0: 

199 return "OTHER_MEANING", 0.5 

200 other_ratio: Final = other_count / total 

201 comp_ratio: Final = comp_count / total 

202 if other_ratio >= 0.6: 

203 return "OTHER_MEANING", min(0.9, 0.5 + 0.4 * other_ratio) 

204 if comp_ratio >= 0.6: 

205 return "COMPETITOR", min(0.9, 0.5 + 0.4 * comp_ratio) 

206 return "OTHER_MEANING", 0.5