""" Domain Term Extractor for RAG Optimization. Extracts domain-specific terms from text chunks and Excel records to enable efficient pre-filtering during vector similarity search. """ import asyncio import json import math import os import re from dataclasses import dataclass, field from typing import Any, Optional, Union from uuid import UUID, uuid4 import structlog from openai import AzureOpenAI, AsyncAzureOpenAI from src.core.config import settings from src.llm import get_async_chat_client, get_chat_client, get_chat_model_name, strip_thinking logger = structlog.get_logger(__name__) # Environment configuration MAX_TERMS_PER_CHUNK = int(os.getenv("DOMAIN_TERM_MAX_PER_CHUNK", "50")) LLM_THRESHOLD = int(os.getenv("DOMAIN_TERM_LLM_THRESHOLD", "2")) ENABLE_CJK_SEGMENTATION = os.getenv("ENABLE_CJK_SEGMENTATION", "true").lower() == "true" ENABLE_PERFORMANCE_MONITORING = os.getenv("ENABLE_PERFORMANCE_MONITORING", "false").lower() == "true" # TEXTIQ-389 confidence constants (SYS-FR-47). RULE_CONFIDENCE = 0.95 LLM_FALLBACK_CONFIDENCE = 0.70 @dataclass class DomainTaxonomy: """Domain taxonomy with categories and term patterns. Defines high-level domain categories and their associated terms for rule-based extraction and category inference. """ categories: dict[str, list[str]] = field(default_factory=lambda: { "finance": [ "revenue", "ebitda", "margin", "profit", "loss", "income", "expense", "balance", "asset", "liability", "equity", "cash flow", "investment", "dividend", "earnings", "debt", "credit", "budget", "forecast", "valuation", "roi", "quarter", "quarterly", "annual", "fiscal", "financial" ], "hr": [ "employee", "salary", "compensation", "benefit", "payroll", "hire", "termination", "performance", "review", "training", "onboarding", "headcount", "workforce", "talent", "recruitment", "retention", "leave", "vacation", "pto", "overtime", "bonus", "promotion" ], "engineering": [ "code", "deploy", "build", "test", "debug", "bug", "feature", "api", "database", "server", "client", "frontend", "backend", "infrastructure", "architecture", "design", "implementation", "sprint", "ticket", "release", "version", "commit", "branch", "merge", "pull request", "ci/cd" ], "legal": [ "contract", "agreement", "clause", "term", "condition", "compliance", "regulation", "law", "statute", "liability", "indemnity", "warranty", "intellectual property", "patent", "trademark", "copyright", "license", "litigation", "dispute", "settlement", "amendment", "exhibit" ], "operations": [ "process", "workflow", "procedure", "supply chain", "inventory", "logistics", "vendor", "supplier", "procurement", "fulfillment", "delivery", "shipment", "warehouse", "distribution", "capacity", "throughput", "efficiency", "quality", "sla", "kpi" ], "marketing": [ "campaign", "brand", "customer", "lead", "conversion", "engagement", "analytics", "traffic", "channel", "social media", "content", "seo", "sem", "advertising", "promotion", "funnel", "segment", "target", "awareness", "retention", "churn" ] }) def get_all_terms(self) -> set[str]: """Get all terms across all categories.""" all_terms = set() for terms in self.categories.values(): all_terms.update(term.lower() for term in terms) return all_terms def infer_category(self, terms: list[str]) -> str: """Infer domain category from extracted terms. Args: terms: List of extracted terms Returns: Category name with highest term match count, or "general" if no clear match """ if not terms: return "general" terms_lower = {term.lower() for term in terms} category_scores = {} for category, category_terms in self.categories.items(): category_terms_lower = {term.lower() for term in category_terms} matches = len(terms_lower & category_terms_lower) if matches > 0: category_scores[category] = matches if not category_scores: return "general" # Return category with highest score return max(category_scores.items(), key=lambda x: x[1])[0] # LLM prompt for term extraction TERM_EXTRACTION_PROMPT = """Extract domain-specific terms from the following text. Focus on: - Business domain keywords (finance, HR, legal, engineering, etc.) - Technical terminology specific to the content - Important entities and concepts - Industry-specific jargon Rules: - Extract 3-15 terms (or fewer if content is sparse) - Use lowercase - Include both single words and short phrases (max 3 words) - Avoid generic words like "the", "is", "and" - Focus on nouns and noun phrases Text: {content} Output ONLY a JSON array of terms: ["term1", "term2", "term3"] """ class DomainTermExtractor: """Extracts domain-specific terms from chunks for RAG pre-filtering.""" def __init__( self, taxonomy: Optional[DomainTaxonomy] = None, client: Optional[AzureOpenAI] = None ): """Initialize the domain term extractor. Args: taxonomy: DomainTaxonomy instance (uses default if None) client: Optional AzureOpenAI client (lazy-loaded if None) """ self.taxonomy = taxonomy or DomainTaxonomy() self._client = client self._client_initialized = client is not None self.logger = structlog.get_logger(__name__) self._sudachi_tokenizer = None self._sudachi_init_attempted = False def _get_client(self): """Get or create LLM client lazily. Lazy initialization prevents creating HTTP clients before process forking, which causes issues in Celery/Airflow workers. """ if not self._client_initialized: self._client = get_chat_client() self._client_initialized = True return self._client def extract_from_chunk( self, content: str, metadata: Optional[dict[str, Any]] = None ) -> dict[str, Any]: """Extract domain terms from a text chunk. Uses hybrid approach: 1. Rule-based extraction (fast keyword matching) 2. LLM enhancement (for complex/ambiguous content) Args: content: Text content of the chunk metadata: Optional metadata dict (e.g., {"page_number": 1}) Returns: Dict with: - domain_terms: List of extracted terms (max MAX_TERMS_PER_CHUNK) - domain_category: Inferred category string - extraction_method: "rules" or "llm" Example: >>> extractor = DomainTermExtractor() >>> result = extractor.extract_from_chunk("Q4 EBITDA increased...") >>> result {"domain_terms": ["ebitda", "q4", "revenue"], "domain_category": "finance", "extraction_method": "rules"} """ if not content or not content.strip(): return { "domain_terms": [], "domain_category": "general", "extraction_method": "none" } # Try rule-based extraction first rule_terms = self._extract_by_rules(content) terms: list[str] = list(rule_terms) confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms} extraction_method = "rules" # Fall back to LLM if too few terms found if len(terms) < LLM_THRESHOLD: self.logger.info( "using_llm_extraction", terms_found=len(terms), threshold=LLM_THRESHOLD ) for llm_term, llm_conf in self._llm_extract(content): terms.append(llm_term) # Max across sources (rule=0.95 vs LLM-computed). confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf) extraction_method = "llm" # Deduplicate (preserving order) and limit terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK] # Infer category category = self.taxonomy.infer_category(terms) return { "domain_terms": terms, "domain_category": category, "domain_confidences": {t: confidences[t] for t in terms}, "extraction_method": extraction_method, } def extract_from_excel_record( self, headers: list[str], row_data: dict[str, Any], sheet_name: Optional[str] = None ) -> dict[str, Any]: """Extract domain terms from an Excel record. Extracts terms from both column headers and cell values. Args: headers: List of column header strings (flat or hierarchical) row_data: Dict mapping headers to cell values sheet_name: Optional sheet name Returns: Dict with domain_terms, domain_category, extraction_method Example: >>> extractor = DomainTermExtractor() >>> result = extractor.extract_from_excel_record( ... headers=["Employee ID", "Salary", "Department"], ... row_data={"Employee ID": "E123", "Salary": 75000, "Department": "HR"} ... ) >>> result["domain_category"] "hr" """ # Combine headers and values into searchable text content_parts = [] # Add sheet name if available if sheet_name: content_parts.append(sheet_name) # Add all headers content_parts.extend(headers) # Add string values (skip numeric values for term extraction) for value in row_data.values(): if isinstance(value, str) and value.strip(): content_parts.append(value) combined_content = " ".join(content_parts) # Use standard chunk extraction return self.extract_from_chunk(combined_content) def _extract_by_rules(self, content: str) -> list[str]: """Extract terms using rule-based keyword matching. Fast pattern matching against taxonomy terms using fuzzy matching. Args: content: Text content Returns: List of matched terms """ content_lower = content.lower() matched_terms = [] # Match against all taxonomy terms for term in self.taxonomy.get_all_terms(): # Use word boundary matching to avoid partial matches pattern = r'\b' + re.escape(term) + r'\b' if re.search(pattern, content_lower): matched_terms.append(term) return matched_terms def _get_sudachi_tokenizer(self): """Get or create SudachiPy tokenizer lazily. Lazy initialization prevents creating tokenizer before process forking, which causes issues in Celery/Airflow workers. The _sudachi_init_attempted flag tracks whether initialization has been attempted (regardless of success) to avoid repeated failed attempts. Returns: Tokenizer instance or None if initialization fails """ if not self._sudachi_init_attempted: try: from sudachipy import tokenizer, dictionary tokenizer_obj = dictionary.Dictionary().create() self._sudachi_tokenizer = tokenizer_obj self._sudachi_init_attempted = True except ImportError as e: self.logger.warning( "sudachi_init_failed", error="sudachipy not installed", error_type="ImportError", fallback="segmentation_disabled", action_required="install sudachipy and sudachidict_core packages" ) self._sudachi_init_attempted = True except (FileNotFoundError, PermissionError) as e: self.logger.error( "sudachi_init_failed", error=str(e), error_type=type(e).__name__, fallback="segmentation_disabled", action_required="check dictionary files exist and have correct permissions" ) self._sudachi_init_attempted = True except Exception as e: self.logger.error( "sudachi_init_failed", error=str(e), error_type=type(e).__name__, fallback="segmentation_disabled", action_required="check logs and sudachipy installation" ) self._sudachi_init_attempted = True return self._sudachi_tokenizer def _is_japanese(self, text: str) -> bool: """Detect if text contains Japanese characters. Uses heuristics to distinguish Japanese from Chinese: - Hiragana/Katakana presence strongly indicates Japanese - Kanji-only text with sufficient Hiragana/Katakana ratio is Japanese - Pure Kanji is treated conservatively (could be Chinese) Args: text: Text to analyze Returns: True if Japanese characters detected with high confidence, False otherwise """ if not text: return False # Unicode ranges for Japanese scripts # Hiragana: U+3040-U+309F (unique to Japanese) # Katakana: U+30A0-U+30FF (unique to Japanese) # Kanji: U+4E00-U+9FFF (shared with Chinese) # Check for Hiragana or Katakana - strong indicators of Japanese hiragana_katakana_pattern = r'[\u3040-\u309F\u30A0-\u30FF]' if re.search(hiragana_katakana_pattern, text): return True # If only Kanji present, check if ratio is significant # to avoid false positives on Chinese text kanji_pattern = r'[\u4E00-\u9FFF]' kanji_matches = re.findall(kanji_pattern, text) if kanji_matches: # Only treat as Japanese if Kanji makes up >30% of text # This helps filter out English text with occasional Kanji kanji_ratio = len(kanji_matches) / len(text) return kanji_ratio > 0.3 return False def _segment_japanese(self, text: str) -> str: """Segment Japanese text using SudachiPy. Args: text: Japanese text to segment Returns: Space-separated segmented text, or original text if segmentation fails """ tokenizer_obj = self._get_sudachi_tokenizer() if tokenizer_obj is None: return text # Fallback to original if tokenizer unavailable try: # Use Mode C (long unit) to preserve compound words # Mode is specified here during tokenization, not at tokenizer creation from sudachipy.tokenizer import Tokenizer mode = Tokenizer.SplitMode.C # Measure segmentation performance (if enabled) if ENABLE_PERFORMANCE_MONITORING: import time start_time = time.time() tokens = [m.surface() for m in tokenizer_obj.tokenize(text, mode)] if ENABLE_PERFORMANCE_MONITORING: elapsed_ms = (time.time() - start_time) * 1000 # Validate segmentation output if not tokens or all(len(t.strip()) == 0 for t in tokens): self.logger.warning( "japanese_segmentation_invalid", reason="empty_tokens", fallback="original_text" ) return text # Check if segmentation is meaningful (not all single-char splits) single_char_count = sum(1 for t in tokens if len(t.strip()) == 1) if len(tokens) > 5 and single_char_count / len(tokens) > 0.9: self.logger.warning( "japanese_segmentation_invalid", reason="excessive_single_char_splits", single_char_ratio=round(single_char_count / len(tokens), 2), fallback="original_text" ) return text segmented = " ".join(tokens) # Log success with optional performance metrics log_data = { "char_count": len(text), "token_count": len(tokens) } if ENABLE_PERFORMANCE_MONITORING: log_data["elapsed_ms"] = round(elapsed_ms, 2) self.logger.info("japanese_segmentation_success", **log_data) return segmented except Exception as e: self.logger.warning( "japanese_segmentation_failed", error=str(e), error_type=type(e).__name__, fallback="original_text", action_required="check if error is transient (retry) or indicates tokenizer issue" ) return text # Graceful fallback def _llm_extract(self, content: str) -> list[tuple[str, float]]: """Extract terms using LLM for complex/ambiguous content. Returns (term, confidence) pairs. Confidence is `exp(min(token_logprobs))` over the tokens spanning each JSON-quoted term in the response, with ``LLM_FALLBACK_CONFIDENCE`` (0.70) on any alignment failure. """ import time result_text = "" try: client = self._get_client() # Truncate content if too long (max ~2000 tokens) # Do this BEFORE segmentation to avoid expansion issues if len(content) > 8000: content = content[:8000] + "..." # PREPROCESSING: Segment Japanese text if enabled if ENABLE_CJK_SEGMENTATION: if self._is_japanese(content): self.logger.info( "japanese_text_detected", char_count=len(content) ) content = self._segment_japanese(content) else: self.logger.debug("cjk_segmentation_disabled", reason="env_var_false") prompt = TERM_EXTRACTION_PROMPT.format(content=content) # Add rate limiting: small delay between API calls to avoid overwhelming the service time.sleep(0.1) # 100ms delay between calls start_time = time.time() response = client.chat.completions.create( model=get_chat_model_name(), messages=[ {"role": "system", "content": "You are a domain term extraction expert."}, {"role": "user", "content": prompt} ], temperature=0.0, max_tokens=500, logprobs=True, timeout=30.0 # 30 second timeout per call ) elapsed = time.time() - start_time if elapsed > 10.0: self.logger.warning("llm_call_slow", elapsed_seconds=round(elapsed, 2)) result_text = strip_thinking(response.choices[0].message.content).strip() # Parse JSON array response (empty after think-stripping = no terms found) if not result_text: self.logger.warning("llm_empty_after_think_strip") return [] raw_terms = json.loads(result_text) if not isinstance(raw_terms, list): self.logger.warning( "llm_returned_non_list", result=result_text, action_required="verify prompt is correct; may be transient LLM issue - retry recommended" ) return [] original_terms = [t for t in raw_terms if isinstance(t, str) and t] confidences = self._confidence_from_logprobs(response, original_terms) scored = [(t.lower(), confidences.get(t.lower(), LLM_FALLBACK_CONFIDENCE)) for t in original_terms] self.logger.info("llm_extraction_success", terms_count=len(scored)) return scored except json.JSONDecodeError as e: self.logger.error( "llm_json_parse_error", error=str(e), result=result_text, action_required="likely transient LLM formatting issue - retry recommended; if persistent, check prompt engineering" ) return [] except Exception as e: error_type = type(e).__name__ # Categorize error types for actionable guidance if "Auth" in error_type or "Permission" in error_type or "API" in error_type: action = "check Azure OpenAI credentials and endpoint configuration" likely_transient = False elif "Timeout" in error_type or "Connection" in error_type: action = "network or service issue - retry recommended" likely_transient = True else: action = "investigate error type and context; check logs for details" likely_transient = None self.logger.error( "llm_extraction_failed", error=str(e), error_type=error_type, likely_transient=likely_transient, action_required=action ) return [] def _confidence_from_logprobs( self, response: Any, terms: list[str], ) -> dict[str, float]: """Compute per-term confidence from chat-completion logprobs. Aggregation is weakest-link: ``exp(min(token_logprobs))`` over the tokens that span each JSON-quoted term in the response. Missing logprobs or unalignable terms fall back to ``LLM_FALLBACK_CONFIDENCE`` (0.70). Returns a dict keyed by the **lowercased** term. """ fallback = {t.lower(): LLM_FALLBACK_CONFIDENCE for t in terms} try: lp = response.choices[0].logprobs content_tokens = getattr(lp, "content", None) if lp is not None else None if not content_tokens: self.logger.warning("llm_logprobs_missing", fallback=LLM_FALLBACK_CONFIDENCE) return fallback # Build concatenated text + char-offset → token-index mapping. parts: list[str] = [] logprob_values: list[float] = [] char_to_token: list[int] = [] for idx, tok in enumerate(content_tokens): text = getattr(tok, "token", "") or "" if not isinstance(text, str): return fallback parts.append(text) logprob_values.append(float(getattr(tok, "logprob", 0.0))) char_to_token.extend([idx] * len(text)) full_lower = "".join(parts).lower() out: dict[str, float] = {} for term in terms: term_lower = term.lower() needle = f'"{term_lower}"' pos = full_lower.find(needle) if pos < 0: self.logger.warning( "llm_logprob_term_unalignable", term=term_lower, fallback=LLM_FALLBACK_CONFIDENCE, ) out[term_lower] = LLM_FALLBACK_CONFIDENCE continue # Skip opening quote; end_char points at closing quote (exclusive). start_char = pos + 1 end_char = pos + len(needle) - 1 if start_char >= len(char_to_token) or end_char <= start_char: out[term_lower] = LLM_FALLBACK_CONFIDENCE continue start_tok = char_to_token[start_char] end_tok = char_to_token[end_char - 1] + 1 slice_ = logprob_values[start_tok:end_tok] if not slice_: out[term_lower] = LLM_FALLBACK_CONFIDENCE continue conf = math.exp(min(slice_)) out[term_lower] = max(0.0, min(1.0, conf)) return out except (AttributeError, TypeError, ValueError, IndexError): return fallback def _get_async_client(self): """Get or create async LLM client lazily.""" if not hasattr(self, '_async_client') or self._async_client is None: self._async_client = get_async_chat_client() return self._async_client async def _llm_extract_async(self, content: str) -> list[tuple[str, float]]: """Async version of _llm_extract. Returns (term, confidence) pairs.""" try: client = self._get_async_client() if len(content) > 8000: content = content[:8000] + "..." if ENABLE_CJK_SEGMENTATION and self._is_japanese(content): content = self._segment_japanese(content) prompt = TERM_EXTRACTION_PROMPT.format(content=content) response = await asyncio.wait_for( client.chat.completions.create( model=get_chat_model_name(), messages=[ {"role": "system", "content": "You are a domain term extraction expert."}, {"role": "user", "content": prompt} ], temperature=0.0, max_tokens=500, logprobs=True, ), timeout=30.0, ) result_text = strip_thinking(response.choices[0].message.content).strip() if not result_text: return [] raw_terms = json.loads(result_text) if not isinstance(raw_terms, list): return [] original_terms = [t for t in raw_terms if isinstance(t, str) and t] confidences = self._confidence_from_logprobs(response, original_terms) return [(t.lower(), confidences.get(t.lower(), LLM_FALLBACK_CONFIDENCE)) for t in original_terms] except Exception as e: self.logger.warning( "async_llm_extraction_failed", error=str(e), error_type=type(e).__name__, ) return [] async def extract_from_chunks_async( self, chunks: list[dict[str, Any]], doc_type: str = "excel", ) -> list[dict[str, Any]]: """Extract domain terms from multiple chunks concurrently. Uses rule-based extraction first, then dispatches LLM fallback calls concurrently for chunks that need it. Args: chunks: List of chunk dicts doc_type: Document type ("excel" or text type) Returns: List of dicts with domain_category and domain_terms per chunk """ concurrency = settings.llm_concurrency semaphore = asyncio.Semaphore(concurrency) self.logger.info( "async_domain_extraction_starting", total_chunks=len(chunks), concurrency=concurrency, ) async def extract_single(idx: int, chunk: dict) -> dict[str, Any]: """Extract domain terms from a single chunk.""" try: if doc_type == "excel": content = chunk.get("content", {}) if isinstance(content, dict): headers = list(content.keys()) sheet_name = chunk.get("_source", {}).get("sheet", "") content_parts = [] if sheet_name: content_parts.append(sheet_name) content_parts.extend(headers) for v in content.values(): if isinstance(v, str) and v.strip(): content_parts.append(v) combined = " ".join(content_parts) else: combined = str(content) else: text = chunk.get("text") or chunk.get("content", "") combined = str(text) if not combined or not combined.strip(): return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}} # Rule-based first (fast, no I/O) rule_terms = self._extract_by_rules(combined) terms: list[str] = list(rule_terms) confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms} # LLM fallback if too few terms — with concurrency control if len(terms) < LLM_THRESHOLD: async with semaphore: for llm_term, llm_conf in await self._llm_extract_async(combined): terms.append(llm_term) # Max across sources (rule vs LLM). confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf) terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK] category = self.taxonomy.infer_category(terms) return { "domain_terms": terms, "domain_category": category, "domain_confidences": {t: confidences[t] for t in terms}, } except Exception as e: self.logger.warning( "async_domain_extraction_failed", chunk_idx=idx, error=str(e), ) return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}} import logging logging.getLogger("httpx").setLevel(logging.WARNING) from tqdm.asyncio import tqdm_asyncio tasks = [extract_single(idx, chunk) for idx, chunk in enumerate(chunks)] results = await tqdm_asyncio.gather( *tasks, desc="Extracting domain terms" ) # Log category distribution category_counts: dict[str, int] = {} for data in results: cat = data["domain_category"] category_counts[cat] = category_counts.get(cat, 0) + 1 self.logger.info( "async_domain_extraction_complete", total_chunks=len(chunks), categories=category_counts, ) return list(results) async def extract_from_nodes_async( self, nodes: list, doc_type: str = "word", ) -> tuple[set, dict[str, int], dict[str, float]]: """Extract domain terms from LlamaIndex nodes concurrently. Mutates node.metadata in place with domain_category and domain_terms. Uses rule-based extraction first, then dispatches LLM fallback calls concurrently for nodes that need it. Args: nodes: List of LlamaIndex TextNode/BaseNode objects doc_type: Document type ("excel", "word", "pdf", etc.) Returns: Tuple of (all_terms set, category_counts dict, term_confidences dict). ``term_confidences`` holds the per-term max confidence across nodes. """ from collections import Counter concurrency = settings.llm_concurrency semaphore = asyncio.Semaphore(concurrency) self.logger.info( "async_node_domain_extraction_starting", total_nodes=len(nodes), concurrency=concurrency, ) async def extract_single(idx: int, node) -> dict[str, Any]: """Extract domain terms from a single node.""" try: text = node.text if hasattr(node, "text") else str(node) metadata = node.metadata if hasattr(node, "metadata") else {} if not text or not text.strip(): return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}} if doc_type == "excel": sheet_name = metadata.get("sheet_name", "") combined = f"{sheet_name} {text}" if sheet_name else text else: combined = text # Rule-based first (fast, no I/O) rule_terms = self._extract_by_rules(combined) terms: list[str] = list(rule_terms) confidences: dict[str, float] = {t: RULE_CONFIDENCE for t in rule_terms} # LLM fallback if too few terms — with concurrency control if len(terms) < LLM_THRESHOLD: async with semaphore: for llm_term, llm_conf in await self._llm_extract_async(combined): terms.append(llm_term) # Max across sources (rule vs LLM). confidences[llm_term] = max(confidences.get(llm_term, 0.0), llm_conf) terms = list(dict.fromkeys(terms))[:MAX_TERMS_PER_CHUNK] category = self.taxonomy.infer_category(terms) return { "domain_terms": terms, "domain_category": category, "domain_confidences": {t: confidences[t] for t in terms}, } except Exception as e: self.logger.warning( "async_node_domain_extraction_failed", node_idx=idx, error=str(e), ) return {"domain_terms": [], "domain_category": "general", "domain_confidences": {}} import logging logging.getLogger("httpx").setLevel(logging.WARNING) from tqdm.asyncio import tqdm_asyncio tasks = [extract_single(idx, node) for idx, node in enumerate(nodes)] results = await tqdm_asyncio.gather( *tasks, desc="Extracting domain terms from nodes" ) # Apply results to node metadata in place all_terms: set = set() category_counts = Counter() term_confidences: dict[str, float] = {} for node, result in zip(nodes, results): node.metadata["domain_category"] = result["domain_category"] node.metadata["domain_terms"] = result["domain_terms"] all_terms.update(result["domain_terms"]) category_counts[result["domain_category"]] += 1 for t, c in result.get("domain_confidences", {}).items(): # Max across nodes (same term appearing in multiple chunks). term_confidences[t] = max(term_confidences.get(t, 0.0), c) self.logger.info( "async_node_domain_extraction_complete", total_nodes=len(nodes), categories=dict(category_counts), ) return all_terms, dict(category_counts), term_confidences def detect_language(self, term: str) -> str: """Detect language of term using Unicode ranges. Args: term: Term to detect language for Returns: Language code: "ja", "vi", or "en" """ # Check for Japanese characters (Hiragana, Katakana, CJK) if self._is_japanese(term): return "ja" # Check for Vietnamese diacritics vietnamese_chars = set("àáảãạăằắẳẵặâầấẩẫậèéẻẽẹêềếểễệìíỉĩịòóỏõọôồốổỗộơờớởỡợùúủũụưừứửữựỳýỷỹỵđ") if any(c in vietnamese_chars for c in term.lower()): return "vi" return "en" # Default @staticmethod def _resolve_persist_confidence( term: str, confidence_by_term: Optional[dict[str, float]], ) -> float: """Per-row confidence lookup for ``persist_new_terms``. Legacy callers pass ``confidence_by_term=None`` → every row defaults to ``LLM_FALLBACK_CONFIDENCE`` (0.70). Otherwise we look up by lowercased term and clamp to [0, 1] defensively. """ if confidence_by_term is None: return LLM_FALLBACK_CONFIDENCE conf = confidence_by_term.get(term.lower(), LLM_FALLBACK_CONFIDENCE) return max(0.0, min(1.0, float(conf))) def persist_new_terms( self, terms: list[str], domain_category: str, *, tenant_id: Union[str, UUID], node_id: Union[str, UUID], document_id: Optional[Union[str, UUID]] = None, confidence_by_term: Optional[dict[str, float]] = None, ) -> int: """Persist new terms to dictionary with concurrent async translations. Args: terms: List of terms to persist domain_category: Domain category for the terms tenant_id: Owning tenant (required — scopes RLS). node_id: Owning hierarchy node (required — scopes per-node dedup). document_id: Optional UUID or string representation of source document. confidence_by_term: Optional per-term confidence map (lowercased keys). Rows for terms absent from the map (including legacy callers that pass ``None``) default to ``LLM_FALLBACK_CONFIDENCE`` (0.70). Returns: Count of new terms added (0 if all duplicates or on IntegrityError) """ try: from sqlalchemy import create_engine from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from src.db.models import DomainTermDictionary from src.extraction_v2.translation_service import TranslationService from src.knowledge.domain_dictionary_service import DomainDictionaryService except ImportError as e: self.logger.error( "import_error_in_persist_new_terms", error=str(e), missing_module=e.name if hasattr(e, 'name') else "unknown", action_required="ensure_dependencies_in_external_python_venv", hint="SQLAlchemy and application modules must be available in Airflow external_python environment" ) raise sync_db_url = settings.database_url.replace('postgresql+asyncpg://', 'postgresql+psycopg2://') sync_db_url = sync_db_url.replace('@localhost:', '@postgres:') engine = create_engine(sync_db_url) try: with Session(engine) as session: service = DomainDictionaryService( session, tenant_id=str(tenant_id) if tenant_id else None, node_id=str(node_id) if node_id else None, ) translator = TranslationService() existing = service.get_existing_terms_set() # Filter to only new terms new_terms = [t for t in terms if t.lower() not in existing] if not new_terms: return 0 self.logger.info("translating_new_terms", count=len(new_terms)) import logging logging.getLogger("httpx").setLevel(logging.WARNING) from tqdm.asyncio import tqdm_asyncio # Translate all new terms concurrently concurrency = getattr(settings, 'llm_concurrency', 5) semaphore = asyncio.Semaphore(concurrency) async def translate_one(term: str): async with semaphore: source_lang = self.detect_language(term) return await translator.translate_term_async(term, source_lang) async def translate_all(): tasks = [translate_one(t) for t in new_terms] return await tqdm_asyncio.gather( *tasks, desc="Translating domain terms" ) translation_results = asyncio.run(translate_all()) doc_id = None if document_id: doc_id = UUID(document_id) if isinstance(document_id, str) else document_id new_entries = [] for term, (translations, success) in zip(new_terms, translation_results): t_id = UUID(str(tenant_id)) if tenant_id else None n_id = UUID(str(node_id)) if node_id else None conf = self._resolve_persist_confidence(term, confidence_by_term) entry = DomainTermDictionary( id=uuid4(), term_original=term, term_ja=translations["ja"], term_en=translations["en"], term_vi=translations["vi"], source_language=self.detect_language(term), domain_category=domain_category, source_document_id=doc_id, translation_failed=not success, confidence=conf, tenant_id=t_id, node_id=n_id, ) new_entries.append(entry) if new_entries: try: service.add_terms_batch(new_entries) return len(new_entries) except IntegrityError as e: self.logger.warning( "integrity_error_during_batch_insert", error=str(e), attempted_count=len(new_entries), reason="likely_race_condition_duplicate_term" ) return 0 return 0 finally: engine.dispose()