Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/routing/analyzer.py: 21%
81 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Query analyzer for extracting features from queries."""
3from __future__ import annotations
5import re
7from lexigram.ai.rag.multimodal.types import Modality
8from lexigram.ai.rag.routing.types import QueryFeatures, QueryIntent
11class QueryAnalyzer:
12 """Analyzes queries to extract features for routing decisions.
14 Extracts various features from queries including:
15 - Basic features: length, keywords
16 - Intent classification: factual, conversational, analytical, etc.
17 - Language detection
18 - Domain classification
19 - Entity detection
20 - Modality detection
22 Example:
23 ```python
24 analyzer = QueryAnalyzer()
25 features = await analyzer.analyze("How do I configure authentication?")
26 logger.info(f"Intent: {features.intent}")
27 logger.info(f"Keywords: {features.keywords}")
28 logger.info(f"Complexity: {features.complexity}")
29 ```
30 """
32 # Intent patterns (simple keyword-based classification)
33 INTENT_PATTERNS: dict[QueryIntent, list[str]] = {
34 QueryIntent.FACTUAL: [
35 r"\bwhat\b",
36 r"\bwho\b",
37 r"\bwhen\b",
38 r"\bwhere\b",
39 r"\bdefine\b",
40 r"\bexplain\b",
41 r"\btell me\b",
42 ],
43 QueryIntent.PROCEDURAL: [
44 r"\bhow\b",
45 r"\bsteps\b",
46 r"\bguide\b",
47 r"\btutorial\b",
48 r"\bconfigure\b",
49 r"\bsetup\b",
50 r"\binstall\b",
51 ],
52 QueryIntent.ANALYTICAL: [
53 r"\bcompare\b",
54 r"\bdifference\b",
55 r"\bversus\b",
56 r"\bvs\b",
57 r"\banalyze\b",
58 r"\bevaluate\b",
59 r"\bpros and cons\b",
60 ],
61 QueryIntent.CREATIVE: [
62 r"\bwrite\b",
63 r"\bcreate\b",
64 r"\bgenerate\b",
65 r"\bcompose\b",
66 r"\bpoem\b",
67 r"\bstory\b",
68 r"\bideas\b",
69 ],
70 QueryIntent.NAVIGATIONAL: [
71 r"\bfind\b",
72 r"\blocate\b",
73 r"\bpage\b",
74 r"\bdocumentation\b",
75 r"\breference\b",
76 r"\blink\b",
77 ],
78 QueryIntent.CONVERSATIONAL: [
79 r"\bhello\b",
80 r"\bhi\b",
81 r"\bthanks\b",
82 r"\bthank you\b",
83 r"\bhow are you\b",
84 r"\bbye\b",
85 ],
86 }
88 # Domain keywords
89 DOMAIN_KEYWORDS: dict[str, list[str]] = {
90 "technical": [
91 "api",
92 "code",
93 "function",
94 "class",
95 "method",
96 "algorithm",
97 "database",
98 "server",
99 "configuration",
100 "deploy",
101 "debug",
102 ],
103 "medical": [
104 "patient",
105 "disease",
106 "treatment",
107 "symptom",
108 "diagnosis",
109 "medication",
110 "doctor",
111 "hospital",
112 "clinical",
113 ],
114 "legal": [
115 "law",
116 "legal",
117 "contract",
118 "regulation",
119 "compliance",
120 "statute",
121 "court",
122 "litigation",
123 "rights",
124 ],
125 "financial": [
126 "money",
127 "investment",
128 "stock",
129 "profit",
130 "revenue",
131 "cost",
132 "budget",
133 "finance",
134 "accounting",
135 ],
136 }
138 # Stop words to exclude from keywords
139 STOP_WORDS: set[str] = {
140 "a",
141 "an",
142 "and",
143 "are",
144 "as",
145 "at",
146 "be",
147 "by",
148 "for",
149 "from",
150 "has",
151 "he",
152 "in",
153 "is",
154 "it",
155 "its",
156 "of",
157 "on",
158 "that",
159 "the",
160 "to",
161 "was",
162 "will",
163 "with",
164 "this",
165 "but",
166 "they",
167 "have",
168 "had",
169 "what",
170 "when",
171 "where",
172 "who",
173 "which",
174 "why",
175 "how",
176 "or",
177 "can",
178 "could",
179 "should",
180 "would",
181 }
183 def __init__(
184 self,
185 *,
186 extract_keywords: bool = True,
187 detect_entities: bool = True,
188 classify_domain: bool = True,
189 ):
190 """Initialize the query analyzer.
192 Args:
193 extract_keywords: Whether to extract keywords.
194 detect_entities: Whether to detect named entities.
195 classify_domain: Whether to classify query domain.
196 """
197 self.extract_keywords = extract_keywords
198 self.detect_entities = detect_entities
199 self.classify_domain = classify_domain
201 async def analyze(self, query: str) -> QueryFeatures:
202 """Analyze a query and extract features.
204 Args:
205 query: Query text to analyze.
207 Returns:
208 Extracted query features.
209 """
210 # Basic features
211 length = len(query)
212 text_lower = query.lower()
214 # Classify intent
215 intent = self._classify_intent(text_lower)
217 # Extract keywords
218 keywords = self._extract_keywords(query) if self.extract_keywords else []
220 # Detect language (simple heuristic)
221 language = self._detect_language(query)
223 # Classify domain
224 domain = self._classify_domain(text_lower) if self.classify_domain else None
226 # Detect entities (simple pattern-based)
227 has_entities = self._detect_entities(query) if self.detect_entities else False
229 # Detect modalities
230 modalities = self._detect_modalities(text_lower)
232 # Calculate complexity
233 complexity = self._calculate_complexity(query, keywords)
235 return QueryFeatures(
236 text=query,
237 length=length,
238 intent=intent,
239 language=language,
240 domain=domain,
241 keywords=keywords,
242 has_entities=has_entities,
243 modalities=modalities,
244 complexity=complexity,
245 )
247 def _classify_intent(self, query_lower: str) -> QueryIntent:
248 """Classify query intent using pattern matching.
250 Args:
251 query_lower: Lowercased query text.
253 Returns:
254 Classified intent.
255 """
256 scores: dict[QueryIntent, int] = {}
258 for intent, patterns in self.INTENT_PATTERNS.items():
259 score = sum(
260 1
261 for pattern in patterns
262 if re.search(pattern, query_lower, re.IGNORECASE)
263 )
264 if score > 0:
265 scores[intent] = score
267 # Return intent with highest score, default to FACTUAL
268 if scores:
269 return max(scores.items(), key=lambda x: x[1])[0]
270 return QueryIntent.FACTUAL
272 def _extract_keywords(self, query: str) -> list[str]:
273 """Extract keywords from query.
275 Args:
276 query: Query text.
278 Returns:
279 List of extracted keywords.
280 """
281 # Simple word extraction (split on whitespace and punctuation)
282 words = re.findall(r"\b\w+\b", query.lower())
284 # Filter out stop words and short words
285 keywords = [
286 word for word in words if word not in self.STOP_WORDS and len(word) > 2
287 ]
289 # Remove duplicates while preserving order
290 seen = set()
291 unique_keywords = []
292 for keyword in keywords:
293 if keyword not in seen:
294 seen.add(keyword)
295 unique_keywords.append(keyword)
297 return unique_keywords[:10] # Limit to top 10 keywords
299 def _detect_language(self, query: str) -> str:
300 """Detect query language (simple heuristic).
302 Args:
303 query: Query text.
305 Returns:
306 Language code (default: 'en').
307 """
308 # Simple heuristic: check for non-ASCII characters
309 # In production, use langdetect or similar library
310 if any(ord(char) > 127 for char in query):
311 # Non-ASCII detected, could be non-English
312 # For now, still return 'en' as default
313 return "en"
314 return "en"
316 def _classify_domain(self, query_lower: str) -> str | None:
317 """Classify query domain based on keywords.
319 Args:
320 query_lower: Lowercased query text.
322 Returns:
323 Domain classification or None if no match.
324 """
325 scores: dict[str, int] = {}
327 for domain, keywords in self.DOMAIN_KEYWORDS.items():
328 score = sum(1 for keyword in keywords if keyword in query_lower)
329 if score > 0:
330 scores[domain] = score
332 # Return domain with highest score if above threshold
333 if scores:
334 max_domain = max(scores.items(), key=lambda x: x[1])
335 if max_domain[1] >= 2: # Require at least 2 matching keywords
336 return max_domain[0]
338 return None
340 def _detect_entities(self, query: str) -> bool:
341 """Detect if query contains named entities.
343 Args:
344 query: Query text.
346 Returns:
347 True if entities detected, False otherwise.
348 """
349 # Simple pattern-based detection
350 # Look for capitalized words (potential proper nouns)
351 capitalized_words = re.findall(r"\b[A-Z][a-z]+\b", query)
353 # Filter out common sentence starters
354 sentence_starters = {"What", "When", "Where", "Who", "Why", "How", "Which"}
355 entities = list(
356 filter(lambda word: word not in sentence_starters, capitalized_words),
357 )
359 return len(entities) > 0
361 def _detect_modalities(self, query_lower: str) -> list[Modality]:
362 """Detect modalities mentioned in the query.
364 Args:
365 query_lower: Lowercased query text.
367 Returns:
368 List of detected modalities.
369 """
370 modalities = [Modality.TEXT] # Always include text
372 # Check for image-related terms
373 image_terms = [
374 "image",
375 "picture",
376 "photo",
377 "visual",
378 "diagram",
379 "chart",
380 "graph",
381 ]
382 if any(term in query_lower for term in image_terms):
383 modalities.append(Modality.IMAGE)
385 # Check for audio-related terms
386 audio_terms = ["audio", "sound", "music", "voice", "recording", "podcast"]
387 if any(term in query_lower for term in audio_terms):
388 modalities.append(Modality.AUDIO)
390 # Check for video-related terms
391 video_terms = ["video", "movie", "clip", "footage", "film"]
392 if any(term in query_lower for term in video_terms):
393 modalities.append(Modality.VIDEO)
395 return modalities
397 def _calculate_complexity(self, query: str, keywords: list[str]) -> float:
398 """Calculate query complexity score.
400 Args:
401 query: Query text.
402 keywords: Extracted keywords.
404 Returns:
405 Complexity score (0-1).
406 """
407 # Factors:
408 # 1. Query length (longer = more complex)
409 # 2. Number of keywords (more = more complex)
410 # 3. Presence of complex punctuation
411 # 4. Number of clauses (commas, semicolons)
413 # Length score (0-1, max at 500 chars)
414 length_score = min(len(query) / 500, 1.0)
416 # Keyword score (0-1, max at 15 keywords)
417 keyword_score = min(len(keywords) / 15, 1.0)
419 # Punctuation complexity (0-1)
420 complex_punctuation = query.count(",") + query.count(";") + query.count(":")
421 punctuation_score = min(complex_punctuation / 5, 1.0)
423 # Average the scores
424 complexity = (length_score + keyword_score + punctuation_score) / 3
426 return round(complexity, 2)