""" Query Term Matcher for domain-aware retrieval. Extracts domain terms from user queries and matches them against the indexed term set for efficient pre-filtering. """ import json import os import re from typing import Optional import structlog from openai import AzureOpenAI from rapidfuzz import fuzz from src.core.config import settings from src.llm import get_chat_client, get_chat_model_name, strip_thinking from src.extraction_v2.domain_term_extractor import DomainTaxonomy from src.knowledge.term_index import TermIndex logger = structlog.get_logger(__name__) # Configuration MAX_QUERY_TERMS = int(os.getenv("QUERY_TERM_MAX_RESULTS", "10")) LLM_EXPANSION_ENABLED = os.getenv("QUERY_TERM_LLM_EXPANSION", "true").lower() == "true" FUZZY_MATCH_THRESHOLD = 85 # Minimum fuzzy match score (0-100) # LLM prompt for query term expansion QUERY_TERM_EXPANSION_PROMPT = """Extract domain-specific search terms from this user query. The user is searching through business documents. Identify key domain terms that would help filter relevant documents before vector search. Query: {query} Tasks: 1. Extract explicit domain terms from the query 2. Add semantically related terms that might appear in relevant documents 3. Focus on business domain keywords (finance, HR, operations, etc.) Rules: - Return 3-10 terms maximum - Use lowercase - Include both exact terms and close synonyms - Avoid generic stopwords Output ONLY a JSON object: {{ "terms": ["term1", "term2", ...], "category": "finance" | "hr" | "engineering" | "legal" | "operations" | "marketing" | "general" }} """ class QueryTermMatcher: """Matches query terms against indexed domain terms for pre-filtering. Uses hybrid approach: 1. Fast keyword matching (exact + fuzzy) 2. Optional LLM semantic expansion for complex queries """ def __init__( self, term_index: Optional[TermIndex] = None, taxonomy: Optional[DomainTaxonomy] = None, client: Optional[AzureOpenAI] = None, ): """Initialize query term matcher. Args: term_index: TermIndex instance for cached term lookups taxonomy: DomainTaxonomy for category inference client: Optional AzureOpenAI client for LLM expansion """ self.term_index = term_index or TermIndex() self.taxonomy = taxonomy or DomainTaxonomy() self._client = client self._client_initialized = client is not None self.logger = structlog.get_logger(__name__) def _get_client(self): """Get or create LLM client lazily.""" if not self._client_initialized: self._client = get_chat_client() self._client_initialized = True return self._client async def extract_query_terms( self, query: str, tenant_id: str = "default" ) -> dict: """Extract domain terms from user query. Uses keyword matching first, with optional LLM expansion for queries with few matched terms. Args: query: User query string tenant_id: Tenant identifier for term index lookup Returns: Dict with: - matched_terms: List of matched domain terms - domain_category: Inferred category - matching_method: "keyword" or "llm" Example: >>> matcher = QueryTermMatcher() >>> result = await matcher.extract_query_terms("What is Q4 EBITDA?") >>> result {"matched_terms": ["ebitda", "quarter"], "domain_category": "finance", "matching_method": "keyword"} """ if not query or not query.strip(): return { "matched_terms": [], "domain_category": "general", "matching_method": "none" } # Try keyword matching first matched_terms = await self._keyword_match(query, tenant_id) matching_method = "keyword" # Fall back to LLM expansion if enabled and few terms found if LLM_EXPANSION_ENABLED and len(matched_terms) < 2: self.logger.info( "using_llm_expansion", matched_terms=len(matched_terms), query=query[:100] ) llm_result = self._llm_expand(query) matched_terms.extend(llm_result.get("terms", [])) matching_method = "llm" # Deduplicate and limit matched_terms = list(dict.fromkeys(matched_terms))[:MAX_QUERY_TERMS] # Infer category from matched terms category = self.taxonomy.infer_category(matched_terms) return { "matched_terms": matched_terms, "domain_category": category, "matching_method": matching_method } async def _keyword_match(self, query: str, tenant_id: str) -> list[str]: """Match query keywords against taxonomy and indexed terms. Uses both exact matching and fuzzy matching for robustness. Args: query: User query string tenant_id: Tenant identifier Returns: List of matched terms """ query_lower = query.lower() matched_terms = [] # 1. Exact match against taxonomy terms for term in self.taxonomy.get_all_terms(): pattern = r'\b' + re.escape(term) + r'\b' if re.search(pattern, query_lower): matched_terms.append(term) # 2. Fuzzy match against indexed terms (if available) indexed_terms = await self.term_index.load_tenant_terms(tenant_id) if indexed_terms: # Extract words from query for fuzzy matching query_words = re.findall(r'\b\w+\b', query_lower) for term in indexed_terms: # Skip if already matched exactly if term in matched_terms: continue # Fuzzy match against query words for word in query_words: score = fuzz.ratio(term, word) if score >= FUZZY_MATCH_THRESHOLD: matched_terms.append(term) break return matched_terms def _llm_expand(self, query: str) -> dict: """Expand query terms using LLM for semantic understanding. Args: query: User query string Returns: Dict with "terms" list and "category" string """ try: client = self._get_client() # Truncate very long queries if len(query) > 500: query = query[:500] + "..." prompt = QUERY_TERM_EXPANSION_PROMPT.format(query=query) response = client.chat.completions.create( model=get_chat_model_name(), messages=[ {"role": "system", "content": "You are a query analysis expert."}, {"role": "user", "content": prompt} ], temperature=0.0, max_tokens=300 ) result_text = strip_thinking(response.choices[0].message.content).strip() if not result_text: return {} # Parse JSON response result = json.loads(result_text) if not isinstance(result, dict): self.logger.warning("llm_returned_non_dict", result=result_text) return {"terms": [], "category": "general"} # Normalize terms to lowercase terms = [term.lower() for term in result.get("terms", [])] self.logger.info( "llm_expansion_success", terms_count=len(terms), category=result.get("category", "general") ) return { "terms": terms, "category": result.get("category", "general") } except json.JSONDecodeError as e: self.logger.error("llm_json_parse_error", error=str(e), result=result_text) return {"terms": [], "category": "general"} except Exception as e: self.logger.error("llm_expansion_failed", error=str(e)) return {"terms": [], "category": "general"} def _infer_category_from_terms(self, terms: list[str]) -> str: """Infer domain category from matched terms. Args: terms: List of matched domain terms Returns: Category name or "general" """ return self.taxonomy.infer_category(terms)