""" Dynamic Content Detector for Cache Optimization. This module provides a scalable, language-agnostic approach to detecting dynamic content in prompts. Dynamic content (dates, prices, user data, session info) breaks cache prefixes. By detecting and moving dynamic content to the end, we maximize cache hits. Design Philosophy: - NO HARDCODED PATTERNS for locale-specific content (no month names, etc.) - Structural detection: "Label: value" patterns where LABEL indicates dynamism - Entropy-based detection: High entropy = likely dynamic (UUIDs, tokens, hashes) - Universal patterns only: ISO 8601, UUIDs, Unix timestamps (truly universal) Tiers (configurable, each adds latency): Tier 1: Regex (~0ms) - Structural patterns, universal formats, entropy-based Tier 2: NER (~5-10ms) - Named Entity Recognition for names, money, orgs Tier 3: Semantic (~20-50ms) - Embedding similarity to known dynamic patterns Usage: from headroom.cache.dynamic_detector import DynamicContentDetector detector = DynamicContentDetector(tiers=["regex", "ner"]) result = detector.detect("Session: abc123. User: John paid $500.") # result.spans = [ # DynamicSpan(text="Session: abc123", category="session", tier="regex", ...), # DynamicSpan(text="John", category="person", tier="ner", ...), # DynamicSpan(text="$500", category="money", tier="ner", ...), # ] """ from __future__ import annotations import math import re from dataclasses import dataclass, field from enum import Enum from importlib.util import find_spec from typing import Any, Literal from headroom.models.config import ML_MODEL_DEFAULTS # Optional ML dependencies are checked without importing them so this module # stays cheap to import during proxy startup. _SPACY_AVAILABLE = find_spec("spacy") is not None _SENTENCE_TRANSFORMERS_AVAILABLE = ( find_spec("numpy") is not None and find_spec("sentence_transformers") is not None ) class DynamicCategory(str, Enum): """Categories of dynamic content.""" # Tier 1: Structural/Regex detectable DATE = "date" TIME = "time" DATETIME = "datetime" TIMESTAMP = "timestamp" UUID = "uuid" REQUEST_ID = "request_id" VERSION = "version" SESSION = "session" USER_DATA = "user_data" IDENTIFIER = "identifier" # Generic high-entropy ID # Tier 2: NER detectable PERSON = "person" MONEY = "money" ORG = "org" LOCATION = "location" # Tier 3: Semantic VOLATILE = "volatile" # Semantically detected as changing REALTIME = "realtime" # Fallback UNKNOWN = "unknown" @dataclass class DynamicSpan: """A span of dynamic content detected in text.""" # The actual text matched text: str # Position in original content start: int end: int # What category of dynamic content category: DynamicCategory # Which tier detected it tier: Literal["regex", "ner", "semantic"] # Confidence score (0-1) confidence: float = 1.0 # Additional metadata (pattern name, entity type, etc.) metadata: dict[str, Any] = field(default_factory=dict) @dataclass class DetectionResult: """Result of dynamic content detection.""" # All detected spans spans: list[DynamicSpan] # Content with dynamic parts removed static_content: str # Content that was extracted (for reinsertion at end) dynamic_content: str # Which tiers were used tiers_used: list[str] # Processing time in milliseconds processing_time_ms: float = 0.0 # Any warnings (e.g., "spaCy not available, skipping NER") warnings: list[str] = field(default_factory=list) @dataclass class DetectorConfig: """Configuration for the dynamic content detector.""" # Which tiers to enable (order matters - later tiers can use earlier results) tiers: list[Literal["regex", "ner", "semantic"]] = field(default_factory=lambda: ["regex"]) # Tier 1: Structural labels that indicate dynamic content # These are the KEY names that hint the VALUE is dynamic # Users can add domain-specific labels dynamic_labels: list[str] = field( default_factory=lambda: [ # Time-related "date", "time", "timestamp", "datetime", "created", "updated", "modified", "expires", "last", "current", "today", "now", # Identifiers "id", "uuid", "guid", "session", "request", "trace", "span", "transaction", "correlation", "token", "key", "secret", # User-related "user", "username", "email", "name", "phone", "address", "customer", "client", "employee", "member", # System state "version", "build", "commit", "branch", "revision", "status", "state", "count", "total", "balance", "remaining", "load", "queue", "active", "pending", # Order/ticket related "order", "ticket", "case", "invoice", "reference", ] ) # Tier 1: Custom regex patterns (user-provided) custom_patterns: list[tuple[str, DynamicCategory]] = field(default_factory=list) # Entropy threshold for detecting random strings (0-1 scale normalized) # Higher = more selective (only very random strings) entropy_threshold: float = 0.7 # Minimum length for entropy-based detection min_entropy_length: int = 8 # Tier 2: NER config spacy_model: str = field(default_factory=lambda: ML_MODEL_DEFAULTS.spacy) ner_entity_types: list[str] = field( default_factory=lambda: ["DATE", "TIME", "MONEY", "PERSON", "ORG", "GPE"] ) # Tier 3: Semantic config embedding_model: str = field(default_factory=lambda: ML_MODEL_DEFAULTS.sentence_transformer) semantic_threshold: float = 0.7 # General min_span_length: int = 2 merge_overlapping: bool = True def calculate_entropy(s: str) -> float: """ Calculate Shannon entropy of a string, normalized to 0-1. Higher entropy = more random/unpredictable = likely dynamic. - "aaaaaaa" -> ~0 (low entropy, predictable) - "a1b2c3d4" -> ~0.7 (medium entropy) - "550e8400-e29b-41d4" -> ~0.9 (high entropy, random-looking) Returns: Normalized entropy (0-1). Higher = more likely dynamic. """ if not s: return 0.0 # Count character frequencies freq: dict[str, int] = {} for char in s: freq[char] = freq.get(char, 0) + 1 # Calculate entropy length = len(s) entropy = 0.0 for count in freq.values(): p = count / length entropy -= p * math.log2(p) # Normalize: max entropy for string of length n with k unique chars # is log2(min(n, alphabet_size)). We'll normalize by log2(length) # to get a 0-1 scale max_entropy = math.log2(length) if length > 1 else 1.0 return entropy / max_entropy if max_entropy > 0 else 0.0 class RegexDetector: """ Tier 1: Scalable pattern detection. Uses THREE strategies (no hardcoded month names!): 1. Structural: "Label: value" patterns where label indicates dynamic content 2. Universal: Truly universal formats (ISO 8601, UUID, Unix timestamps) 3. Entropy: High-entropy strings (tokens, hashes, IDs) """ # Universal patterns (these formats are language-agnostic) UNIVERSAL_PATTERNS = [ # UUID - truly universal format ( r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", DynamicCategory.UUID, "uuid", ), # ISO 8601 datetime (most universal date format) ( r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?", DynamicCategory.DATETIME, "iso_datetime", ), # ISO 8601 date only (r"\d{4}-\d{2}-\d{2}(?!\d)", DynamicCategory.DATE, "iso_date"), # Unix timestamps (10-13 digits, but NOT within longer numbers) (r"(?(?:{labels_pattern}))(?P\s*[:=]\s*|\s+)(?P[^\n,;]+)", re.IGNORECASE, ) # Compile custom patterns self._custom_patterns: list[tuple[re.Pattern[str], DynamicCategory]] = [ (re.compile(pattern, re.IGNORECASE), category) for pattern, category in config.custom_patterns ] def detect(self, content: str) -> list[DynamicSpan]: """Detect dynamic content using structural, universal, and entropy detection.""" spans: list[DynamicSpan] = [] seen_ranges: set[tuple[int, int]] = set() # 1. Universal patterns first (most specific) for pattern, category, pattern_name in self._universal_patterns: for match in pattern.finditer(content): start, end = match.start(), match.end() if self._is_overlapping(start, end, seen_ranges): continue if end - start < self.config.min_span_length: continue spans.append( DynamicSpan( text=match.group(), start=start, end=end, category=category, tier="regex", confidence=1.0, metadata={"pattern": pattern_name, "method": "universal"}, ) ) seen_ranges.add((start, end)) # 2. Structural detection: "Label: value" patterns for match in self._structural_pattern.finditer(content): # Get the full match range start, end = match.start(), match.end() # Skip if overlaps with universal patterns if self._is_overlapping(start, end, seen_ranges): continue label = match.group("label").lower() value = match.group("value").strip() # Determine category from label category = self._categorize_label(label) # Only add the value portion (keep label as static) value_start = match.start("value") value_end = match.end("value") # Skip if value is too short or empty if value_end - value_start < self.config.min_span_length: continue if not value.strip(): continue spans.append( DynamicSpan( text=value, start=value_start, end=value_end, category=category, tier="regex", confidence=0.9, metadata={"pattern": "structural", "method": "structural", "label": label}, ) ) seen_ranges.add((value_start, value_end)) # 3. Entropy-based detection for remaining potential IDs spans.extend(self._detect_high_entropy(content, seen_ranges)) # 4. Custom patterns for pattern, category in self._custom_patterns: for match in pattern.finditer(content): start, end = match.start(), match.end() if self._is_overlapping(start, end, seen_ranges): continue if end - start < self.config.min_span_length: continue spans.append( DynamicSpan( text=match.group(), start=start, end=end, category=category, tier="regex", confidence=0.8, metadata={"pattern": "custom", "method": "custom"}, ) ) seen_ranges.add((start, end)) return sorted(spans, key=lambda s: s.start) def _detect_high_entropy( self, content: str, seen_ranges: set[tuple[int, int]], ) -> list[DynamicSpan]: """ Detect high-entropy strings that look like IDs/tokens. Finds alphanumeric sequences and checks their entropy. High entropy = likely random/generated = dynamic. """ spans: list[DynamicSpan] = [] # Find alphanumeric sequences (potential IDs) # Must be at least min_entropy_length chars, mix of letters/numbers pattern = re.compile(r"\b[a-zA-Z0-9_-]{8,}\b") for match in pattern.finditer(content): start, end = match.start(), match.end() text = match.group() # Skip if already detected if self._is_overlapping(start, end, seen_ranges): continue # Skip if too short if len(text) < self.config.min_entropy_length: continue # Skip if all letters or all numbers (not random-looking) if text.isalpha() or text.isdigit(): continue # Skip common words that might look like IDs if text.lower() in {"username", "password", "localhost", "undefined"}: continue # Calculate entropy entropy = calculate_entropy(text) if entropy >= self.config.entropy_threshold: spans.append( DynamicSpan( text=text, start=start, end=end, category=DynamicCategory.IDENTIFIER, tier="regex", confidence=entropy, # Use entropy as confidence metadata={"pattern": "entropy", "method": "entropy", "entropy": entropy}, ) ) seen_ranges.add((start, end)) return spans def _is_overlapping( self, start: int, end: int, seen_ranges: set[tuple[int, int]], ) -> bool: """Check if range overlaps with any existing range.""" return any(not (end <= s or start >= e) for s, e in seen_ranges) def _categorize_label(self, label: str) -> DynamicCategory: """Categorize based on the label name.""" label = label.lower() # Time-related if label in {"date", "datetime", "created", "updated", "modified", "expires", "today"}: return DynamicCategory.DATE if label in {"time", "timestamp", "now"}: return DynamicCategory.TIMESTAMP if label == "current": return DynamicCategory.DATETIME # Identifiers if label in {"id", "uuid", "guid"}: return DynamicCategory.UUID if label in {"session", "request", "trace", "span", "transaction", "correlation"}: return DynamicCategory.SESSION if label in {"token", "key", "secret"}: return DynamicCategory.REQUEST_ID # User-related if label in { "user", "username", "email", "name", "phone", "address", "customer", "client", "employee", "member", }: return DynamicCategory.USER_DATA # System state if label in {"version", "build", "commit", "branch", "revision"}: return DynamicCategory.VERSION if label in { "status", "state", "count", "total", "balance", "remaining", "load", "queue", "active", "pending", }: return DynamicCategory.VOLATILE # Order/ticket if label in {"order", "ticket", "case", "invoice", "reference"}: return DynamicCategory.REQUEST_ID return DynamicCategory.UNKNOWN class NERDetector: """Tier 2: spaCy-based Named Entity Recognition.""" # Map spaCy entity types to our categories ENTITY_MAP = { "DATE": DynamicCategory.DATE, "TIME": DynamicCategory.TIME, "MONEY": DynamicCategory.MONEY, "PERSON": DynamicCategory.PERSON, "ORG": DynamicCategory.ORG, "GPE": DynamicCategory.LOCATION, # Geo-Political Entity "LOC": DynamicCategory.LOCATION, "FAC": DynamicCategory.LOCATION, # Facility "CARDINAL": DynamicCategory.UNKNOWN, # Numbers "ORDINAL": DynamicCategory.UNKNOWN, } def __init__(self, config: DetectorConfig): """Initialize NER detector, loading spaCy model.""" self.config = config self._nlp = None self._load_error: str | None = None if not _SPACY_AVAILABLE: self._load_error = ( "spaCy not installed. Install with: " "pip install spacy && python -m spacy download en_core_web_sm" ) return try: # Use centralized registry for shared model instances from headroom.models.ml_models import MLModelRegistry self._nlp = MLModelRegistry.get_spacy(config.spacy_model) except ImportError: self._load_error = ( "spaCy not installed. Install with: " "pip install spacy && python -m spacy download en_core_web_sm" ) except OSError: self._load_error = ( f"spaCy model '{config.spacy_model}' not found. " f"Install with: python -m spacy download {config.spacy_model}" ) @property def is_available(self) -> bool: """Check if NER is available.""" return self._nlp is not None def detect( self, content: str, existing_spans: list[DynamicSpan] | None = None, ) -> tuple[list[DynamicSpan], str | None]: """ Detect dynamic content using NER. Args: content: Text to analyze. existing_spans: Spans already detected (to avoid duplicates). Returns: Tuple of (new_spans, warning_message). """ if not self.is_available: return [], self._load_error # Get existing ranges to avoid duplicates existing_ranges = set() if existing_spans: existing_ranges = {(s.start, s.end) for s in existing_spans} doc = self._nlp(content) # type: ignore[misc] spans: list[DynamicSpan] = [] for ent in doc.ents: # Skip entity types we don't care about if ent.label_ not in self.config.ner_entity_types: continue # Skip if already detected by regex if (ent.start_char, ent.end_char) in existing_ranges: continue # Check for overlap with existing spans overlaps = any( not (ent.end_char <= s or ent.start_char >= e) for s, e in existing_ranges ) if overlaps: continue # Map to our category category = self.ENTITY_MAP.get(ent.label_, DynamicCategory.UNKNOWN) # Skip unknown categories if category == DynamicCategory.UNKNOWN: continue spans.append( DynamicSpan( text=ent.text, start=ent.start_char, end=ent.end_char, category=category, tier="ner", confidence=0.9, metadata={"entity_type": ent.label_}, ) ) existing_ranges.add((ent.start_char, ent.end_char)) return sorted(spans, key=lambda s: s.start), None class SemanticDetector: """Tier 3: Embedding-based semantic detection.""" # Known phrases that indicate dynamic content # These are SEMANTIC patterns, not literal strings to match DYNAMIC_EXEMPLARS = [ # Time-sensitive "The current date is", "As of today", "Updated on", "Last refreshed", "Real-time data", "Live prices", "Current stock price", # Session-specific "Your session ID", "Your account balance", "Your recent orders", "Your conversation history", # User-specific "Hello [user]", "Dear customer", "Your name is", # System state "Server status", "System load", "Queue length", "Active users", ] def __init__(self, config: DetectorConfig): """Initialize semantic detector with embedding model.""" self.config = config self._model = None self._exemplar_embeddings = None self._load_error: str | None = None if not _SENTENCE_TRANSFORMERS_AVAILABLE: self._load_error = ( "sentence-transformers not installed. " "Install with: pip install sentence-transformers" ) return try: # Use centralized registry for shared model instances from headroom.models.ml_models import MLModelRegistry self._model = MLModelRegistry.get_sentence_transformer(config.embedding_model) # Pre-compute exemplar embeddings self._exemplar_embeddings = self._model.encode( self.DYNAMIC_EXEMPLARS, convert_to_numpy=True, ) except ImportError: self._load_error = ( "sentence-transformers not installed. " "Install with: pip install sentence-transformers" ) except Exception as e: self._load_error = f"Failed to load embedding model: {e}" @property def is_available(self) -> bool: """Check if semantic detection is available.""" return self._model is not None def detect( self, content: str, existing_spans: list[DynamicSpan] | None = None, ) -> tuple[list[DynamicSpan], str | None]: """ Detect dynamic content using semantic similarity. Splits content into sentences and checks each against known dynamic patterns using embedding similarity. Args: content: Text to analyze. existing_spans: Spans already detected (to avoid duplicates). Returns: Tuple of (new_spans, warning_message). """ if not self.is_available: return [], self._load_error # Simple sentence splitting (could use spaCy if available) sentences = self._split_sentences(content) spans: list[DynamicSpan] = [] # Get existing ranges existing_ranges = set() if existing_spans: existing_ranges = {(s.start, s.end) for s in existing_spans} # Encode all sentences if not sentences: return [], None try: import numpy as np except ImportError: return [], "numpy not installed. Install with: pip install numpy" sentence_texts = [s[0] for s in sentences] # `is_available` only guarantees `_model` is set. Guard each piece # separately and *before* encoding so a None never reaches `.T` (a # real crash), mypy can narrow the `Any | None` attributes, and the # caller gets a warning that names the actual missing piece — the # model vs. the exemplar matrix. (Folding both into one guard, as a # prior change did, returned the generic "semantic detector" message # even when only the exemplars were missing.) if self._model is None: return [], self._load_error or "semantic detector is not initialized" if self._exemplar_embeddings is None: return [], "exemplar embeddings not initialized" sentence_embeddings = self._model.encode( sentence_texts, convert_to_numpy=True, ) similarities = np.dot(sentence_embeddings, self._exemplar_embeddings.T) for i, (text, start, end) in enumerate(sentences): # Get max similarity to any exemplar max_sim = float(np.max(similarities[i])) if max_sim < self.config.semantic_threshold: continue # Check overlap with existing spans overlaps = any(not (end <= s or start >= e) for s, e in existing_ranges) if overlaps: continue # Find which exemplar matched best best_exemplar_idx = int(np.argmax(similarities[i])) best_exemplar = self.DYNAMIC_EXEMPLARS[best_exemplar_idx] # Determine category based on exemplar category = self._categorize_exemplar(best_exemplar) spans.append( DynamicSpan( text=text, start=start, end=end, category=category, tier="semantic", confidence=max_sim, metadata={ "matched_exemplar": best_exemplar, "similarity": max_sim, }, ) ) existing_ranges.add((start, end)) return sorted(spans, key=lambda s: s.start), None def _split_sentences(self, content: str) -> list[tuple[str, int, int]]: """Split content into sentences with positions.""" sentences: list[tuple[str, int, int]] = [] pattern = r"[^.!?\n]+[.!?\n]?" for match in re.finditer(pattern, content): text = match.group().strip() if len(text) > 10: sentences.append((text, match.start(), match.end())) return sentences def _categorize_exemplar(self, exemplar: str) -> DynamicCategory: """Categorize based on which exemplar matched.""" exemplar_lower = exemplar.lower() if any(w in exemplar_lower for w in ["date", "today", "updated", "refreshed"]): return DynamicCategory.DATE elif any(w in exemplar_lower for w in ["price", "stock", "live", "real-time"]): return DynamicCategory.REALTIME elif any(w in exemplar_lower for w in ["session", "account", "your"]): return DynamicCategory.SESSION elif any(w in exemplar_lower for w in ["status", "load", "queue", "active"]): return DynamicCategory.VOLATILE else: return DynamicCategory.VOLATILE class DynamicContentDetector: """ Unified dynamic content detector with tiered detection. Key Design Principles: - NO hardcoded locale-specific patterns (no month names) - Structural detection: Labels indicate what's dynamic - Universal patterns: ISO 8601, UUIDs, Unix timestamps - Entropy-based: High entropy = random/generated = dynamic Usage: # Fast mode (regex only - structural + universal + entropy) detector = DynamicContentDetector(DetectorConfig(tiers=["regex"])) # Balanced mode (regex + NER for names/money) detector = DynamicContentDetector(DetectorConfig(tiers=["regex", "ner"])) # Full mode (all tiers) detector = DynamicContentDetector(DetectorConfig( tiers=["regex", "ner", "semantic"] )) result = detector.detect("Session: abc123. User: John paid $500.") """ def __init__(self, config: DetectorConfig | None = None): """Initialize detector with configuration.""" self.config = config or DetectorConfig() # Initialize detectors based on enabled tiers self._regex_detector: RegexDetector | None = None self._ner_detector: NERDetector | None = None self._semantic_detector: SemanticDetector | None = None if "regex" in self.config.tiers: self._regex_detector = RegexDetector(self.config) if "ner" in self.config.tiers: self._ner_detector = NERDetector(self.config) if "semantic" in self.config.tiers: self._semantic_detector = SemanticDetector(self.config) def detect(self, content: str) -> DetectionResult: """ Detect dynamic content in text. Runs enabled tiers in order, accumulating spans. Each tier can see what previous tiers detected. Args: content: Text to analyze. Returns: DetectionResult with spans, static/dynamic content split, etc. """ import time start_time = time.perf_counter() all_spans: list[DynamicSpan] = [] tiers_used: list[str] = [] warnings: list[str] = [] # Tier 1: Regex (structural + universal + entropy) if self._regex_detector: regex_spans = self._regex_detector.detect(content) all_spans.extend(regex_spans) tiers_used.append("regex") # Tier 2: NER if self._ner_detector: ner_spans, ner_warning = self._ner_detector.detect(content, all_spans) all_spans.extend(ner_spans) if ner_warning: warnings.append(ner_warning) elif ner_spans or self._ner_detector.is_available: tiers_used.append("ner") # Tier 3: Semantic if self._semantic_detector: sem_spans, sem_warning = self._semantic_detector.detect(content, all_spans) all_spans.extend(sem_spans) if sem_warning: warnings.append(sem_warning) elif sem_spans or self._semantic_detector.is_available: tiers_used.append("semantic") # Sort by position all_spans = sorted(all_spans, key=lambda s: s.start) # Build static and dynamic content static_content, dynamic_content = self._split_content(content, all_spans) processing_time = (time.perf_counter() - start_time) * 1000 return DetectionResult( spans=all_spans, static_content=static_content, dynamic_content=dynamic_content, tiers_used=tiers_used, processing_time_ms=processing_time, warnings=warnings, ) def _split_content( self, content: str, spans: list[DynamicSpan], ) -> tuple[str, str]: """Split content into static and dynamic parts.""" if not spans: return content, "" static = content dynamic_parts: list[str] = [] for span in reversed(spans): dynamic_parts.append(span.text) static = static[: span.start] + static[span.end :] static = self._clean_static_content(static) dynamic_parts.reverse() dynamic = "\n".join(dynamic_parts) return static, dynamic def _clean_static_content(self, content: str) -> str: """Clean up static content after span removal.""" lines = content.split("\n") cleaned_lines: list[str] = [] prev_blank = False for line in lines: is_blank = not line.strip() if is_blank and prev_blank: continue cleaned_lines.append(line.rstrip()) prev_blank = is_blank return "\n".join(cleaned_lines).strip() @property def available_tiers(self) -> list[str]: """Get list of actually available tiers (dependencies installed).""" available = [] if self._regex_detector: available.append("regex") if self._ner_detector and self._ner_detector.is_available: available.append("ner") if self._semantic_detector and self._semantic_detector.is_available: available.append("semantic") return available # Convenience function def detect_dynamic_content( content: str, tiers: list[Literal["regex", "ner", "semantic"]] | None = None, ) -> DetectionResult: """ Detect dynamic content in text. Convenience function that creates a detector with specified tiers. Args: content: Text to analyze. tiers: Which tiers to use. Default: ["regex"] for speed. Returns: DetectionResult with detected spans and split content. Example: >>> result = detect_dynamic_content( ... "Session: abc123xyz. User: John paid $500.", ... tiers=["regex", "ner"] ... ) >>> print(result.static_content) >>> print(result.dynamic_content) """ config = DetectorConfig(tiers=tiers or ["regex"]) detector = DynamicContentDetector(config) return detector.detect(content)