Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import re | |
| import time | |
| import uuid | |
| from datetime import datetime, timezone | |
| from difflib import SequenceMatcher | |
| from typing import Any, Iterable | |
| TOKEN_RE = re.compile(r"[A-Za-z0-9]+(?:['’-][A-Za-z0-9]+)*") | |
| STOPWORDS = { | |
| "a", "an", "the", "and", "or", "but", "if", "then", "than", | |
| "to", "of", "in", "on", "at", "for", "from", "with", "by", | |
| "as", "is", "am", "are", "was", "were", "be", "been", "being", | |
| "do", "does", "did", "have", "has", "had", "can", "could", "will", | |
| "would", "should", "may", "might", "must", "i", "me", "my", "we", | |
| "us", "our", "you", "your", "he", "him", "his", "she", "her", "they", | |
| "them", "their", "it", "this", "that", "these", "those", "there", "here", | |
| "not", "no", "yes", "very", "just", "also", "so", "too", "about", "into", | |
| "over", "under", "after", "before", "during", "up", "down", "out", "off", | |
| "again", "more", "most", "some", "any", "all", "both", "each", "every", | |
| "which", "what", "when", "where", "who", "why", "how", | |
| } | |
| TEMPORAL_TERMS = { | |
| "today", "tomorrow", "yesterday", "now", "later", "soon", "morning", | |
| "afternoon", "evening", "night", "week", "month", "year", "monday", | |
| "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday", | |
| "january", "february", "march", "april", "may", "june", "july", | |
| "august", "september", "october", "november", "december", "next", "last", | |
| "this", | |
| } | |
| NONLEXICAL_EDIT_TERMS = { | |
| "please", "kindly", "just", "really", "very", "maybe", "perhaps", "actually", | |
| "basically", "simply", "also", "then", "now", "well", | |
| } | |
| COMMON_STYLE_WORDS = { | |
| "good", "great", "nice", "bad", "better", "best", "love", "like", "want", | |
| "need", "think", "feel", "really", "very", "small", "large", "big", "new", | |
| "old", "fast", "slow", "important", "useful", "easy", "hard", "clear", | |
| "simple", "quick", "maybe", "perhaps", "probably", "definitely", "soon", | |
| } | |
| CONTEXT_TAGS = { | |
| "software": { | |
| "software", "code", "coding", "python", "javascript", "typescript", "api", | |
| "sdk", "github", "git", "repository", "repo", "server", "client", "backend", | |
| "frontend", "database", "sql", "docker", "deploy", "deployment", "function", | |
| "class", "library", "framework", "mcp", "langchain", "llm", "model", "ml", | |
| "ai", "embedding", "rag", "agent", "service", "pytorch", "torch", | |
| "grpc", "firestore", "bazel", "pub", "sub", "pipeline", "platform", | |
| "technical", "tool", "tools", "cto", "laptop", "laptops", "debugging", | |
| "workshop", "prototype", "terminal", | |
| }, | |
| "engineering": { | |
| "engineering", "civil", "structural", "column", "columns", "beam", "beams", | |
| "concrete", "steel", "seismic", "earthquake", "foundation", "construction", | |
| "load", "loads", "design", "joint", "slab", "soil", "bridge", "fan", "fans", | |
| "rpm", "motor", "airflow", "air", "delivery", "rjb", "vs30", "opensees", "p-delta", | |
| }, | |
| "work": { | |
| "meeting", "meeting", "client", "customer", "team", "manager", "review", | |
| "deadline", "project", "task", "sprint", "standup", "email", "report", | |
| "presentation", "launch", "roadmap", "action", "followup", "follow", "call", | |
| }, | |
| "food": { | |
| "food", "fruit", "fruits", "grocery", "groceries", "eat", "eating", "lunch", | |
| "dinner", "breakfast", "banana", "bananas", "apple", "apples", "kiwi", "kiwis", | |
| "recipe", "restaurant", "meal", "snack", "cook", "cooking", | |
| }, | |
| "finance": { | |
| "finance", "money", "budget", "salary", "tax", "invoice", "payment", "bank", | |
| "stock", "stocks", "investment", "invest", "revenue", "profit", "expense", | |
| "investor", "investors", "fund", "funding", "portfolio", "partner", "diligence", | |
| "market", "round", "seed", "series", | |
| }, | |
| "personal": { | |
| "personal", "journal", "diary", "home", "family", "friend", "friends", "travel", | |
| "trip", "vacation", "fitness", "gym", "health", "shopping", "buy", "bought", | |
| }, | |
| "education": { | |
| "class", "lecture", "exam", "assignment", "course", "professor", "college", | |
| "university", "study", "jee", "math", "chemistry", "physics", | |
| }, | |
| "physical": { | |
| "wall", "door", "floor", "ceiling", "smudge", "stain", "paint", "mark", | |
| "scratch", "dent", "damage", "damaged", "surface", "object", "paper", | |
| }, | |
| "nature": { | |
| "mountain", "landscape", "forest", "river", "tree", "trees", "weather", | |
| "autumn", "spring", "summer", "winter", "sky", "view", | |
| }, | |
| "shopping": { | |
| "shop", "shopping", "buy", "bought", "purchase", "store", "order", | |
| }, | |
| } | |
| KIND_VALUES = { | |
| "person_name", "organization", "product_service", "project", "place", | |
| "technical_term", "acronym", "identifier", "domain_term", "spelling", "casing", | |
| "other", | |
| } | |
| MAX_EVIDENCE = 16 | |
| MAX_CONTEXTS = 10 | |
| MAX_EXCEPTIONS = 16 | |
| MAX_HISTORY = 50 | |
| ACTIVE_CONFIDENCE = 0.68 | |
| MIN_ACTIVE_POSITIVE = 2 | |
| EXPLICIT_ACTIVE_CONFIDENCE = 0.93 | |
| NEGATIVE_RATIO_DEMOTE = 0.85 | |
| CONTEXT_APPLY_THRESHOLD = 0.60 | |
| CONTEXT_STRONG_THRESHOLD = 0.82 | |
| CONTEXT_NEGATIVE_BLOCK_THRESHOLD = 0.72 | |
| CONTEXT_CONFLICT_THRESHOLD = 0.68 | |
| SPECIFIC_KINDS = { | |
| "organization", | |
| "product_service", | |
| "project", | |
| "technical_term", | |
| "acronym", | |
| "identifier", | |
| "domain_term", | |
| } | |
| ORDINARY_CONFLICT_TAGS = { | |
| "food", | |
| "physical", | |
| "nature", | |
| "shopping", | |
| } | |
| def now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def clamp(value: float, low: float = 0.0, high: float = 0.99) -> float: | |
| try: | |
| value = float(value) | |
| except (TypeError, ValueError): | |
| value = low | |
| return max(low, min(high, value)) | |
| def normalize_text(text: str) -> str: | |
| return re.sub(r"\s+", " ", (text or "").strip()).lower() | |
| def tokenize(text: str) -> list[str]: | |
| return [match.group(0) for match in TOKEN_RE.finditer(text or "")] | |
| def tokenize_with_spans(text: str) -> list[dict[str, Any]]: | |
| return [ | |
| {"text": match.group(0), "start": match.start(), "end": match.end()} | |
| for match in TOKEN_RE.finditer(text or "") | |
| ] | |
| def similarity(a: str, b: str) -> float: | |
| return SequenceMatcher(None, a.lower(), b.lower(), autojunk=False).ratio() | |
| def is_stopword(term: str) -> bool: | |
| return normalize_text(term) in STOPWORDS | |
| def is_capitalized(term: str) -> bool: | |
| letters = [char for char in term if char.isalpha()] | |
| return bool(letters) and letters[0].isupper() | |
| def has_internal_casing(term: str) -> bool: | |
| letters = [char for char in term if char.isalpha()] | |
| return len(letters) >= 2 and any(char.isupper() for char in letters[1:]) | |
| def term_specificity(term: str) -> float: | |
| clean = re.sub(r"[^A-Za-z0-9]", "", term) | |
| if not clean: | |
| return 0.0 | |
| if len(clean) >= 12: | |
| return 0.90 | |
| if len(clean) >= 9: | |
| return 0.78 | |
| if len(clean) >= 6: | |
| return 0.64 | |
| if len(clean) >= 4: | |
| return 0.40 | |
| return 0.15 | |
| def extract_context_tags(text: str, note_title: str = "") -> list[str]: | |
| source = normalize_text(f"{note_title} {text}") | |
| tokens = set(tokenize(source)) | |
| tags = [] | |
| for tag, vocabulary in CONTEXT_TAGS.items(): | |
| hits = len(tokens.intersection(vocabulary)) | |
| if hits: | |
| tags.append(tag) | |
| return tags | |
| def build_scene( | |
| note_title: str, | |
| current_text: str, | |
| input_text: str, | |
| window: int = 600, | |
| ) -> dict[str, Any]: | |
| current_tail = (current_text or "")[-window:] | |
| current_tail = re.split(r"\n\s*\n", current_tail)[-1].strip() | |
| scene_text = re.sub( | |
| r"\s+", | |
| " ", | |
| f"Note: {note_title}. Local context: {current_tail}. Current input: {input_text}", | |
| ).strip() | |
| input_tags = extract_context_tags(input_text, note_title) | |
| local_tags = extract_context_tags(current_tail, note_title) | |
| tags = list(dict.fromkeys([*input_tags, *local_tags]))[:16] | |
| lowered = normalize_text(input_text) | |
| quoted = bool(re.search(r'["“].*?["”]', input_text or "")) | |
| code_like = ( | |
| "```" in (input_text or "") | |
| or ( | |
| bool(re.search(r"\b[a-z_][a-z0-9_]*(?:\([^)]*\))?\b", lowered)) | |
| and "software" in input_tags | |
| ) | |
| ) | |
| return { | |
| "text": scene_text, | |
| "input_text": input_text, | |
| "tags": tags, | |
| "input_tags": input_tags, | |
| "local_tags": local_tags, | |
| "quoted": quoted, | |
| "code_like": code_like, | |
| "note_title": note_title, | |
| } | |
| def context_overlap( | |
| current_scene: str, | |
| note_title: str, | |
| stored_context: str, | |
| stored_tags: list[str] | None = None, | |
| ) -> float: | |
| current_tokens = { | |
| token | |
| for token in tokenize(current_scene) | |
| if len(token) >= 3 and token not in STOPWORDS | |
| } | |
| stored_tokens = { | |
| token | |
| for token in tokenize(stored_context) | |
| if len(token) >= 3 and token not in STOPWORDS | |
| } | |
| if not current_tokens or not stored_tokens: | |
| lexical = 0.0 | |
| else: | |
| intersection = len(current_tokens.intersection(stored_tokens)) | |
| lexical = intersection / max(1, min(len(current_tokens), len(stored_tokens))) | |
| title_bonus = 0.0 | |
| if note_title and stored_context: | |
| title_tokens = set(tokenize(normalize_text(note_title))) | |
| if title_tokens: | |
| title_bonus = len(title_tokens.intersection(stored_tokens)) / max(1, len(title_tokens)) | |
| current_tags = set(extract_context_tags(current_scene, note_title)) | |
| stored_tag_set = set(stored_tags or extract_context_tags(stored_context, "")) | |
| tag_score = len(current_tags.intersection(stored_tag_set)) / max( | |
| 1, len(current_tags.union(stored_tag_set)) | |
| ) if current_tags or stored_tag_set else 0.0 | |
| return clamp( | |
| 0.52 * lexical + 0.18 * title_bonus + 0.30 * tag_score, | |
| 0.0, | |
| 1.0, | |
| ) | |
| def normalize_context_item(item: dict[str, Any]) -> dict[str, Any]: | |
| value = dict(item or {}) | |
| value.setdefault("note_title", "") | |
| value.setdefault("context", "") | |
| value.setdefault("tags", []) | |
| value.setdefault("timestamp", now_iso()) | |
| value.setdefault("strength", 1.0) | |
| value["tags"] = list(dict.fromkeys(str(tag) for tag in value.get("tags", []) if str(tag).strip()))[:12] | |
| value["context"] = str(value.get("context", ""))[:1800] | |
| value["note_title"] = str(value.get("note_title", ""))[:120] | |
| value["strength"] = clamp(value.get("strength", 1.0), 0.1, 1.0) | |
| return value | |
| def normalize_memory(memory: dict[str, Any]) -> dict[str, Any]: | |
| item = dict(memory or {}) | |
| item.setdefault("id", f"mem_{uuid.uuid4().hex[:10]}") | |
| item.setdefault("source", "") | |
| item.setdefault("target", "") | |
| item.setdefault("kind", "other") | |
| item.setdefault("confidence", 0.0) | |
| item.setdefault("evidence_count", 0) | |
| item.setdefault("positive_evidence_count", 0) | |
| item.setdefault("negative_evidence_count", 0) | |
| item.setdefault("explicit_evidence_count", 0) | |
| item.setdefault("status", "candidate") | |
| item.setdefault("created_at", now_iso()) | |
| item.setdefault("updated_at", now_iso()) | |
| item.setdefault("last_used_at", None) | |
| item.setdefault("last_reason", "") | |
| item.setdefault("aliases", []) | |
| item.setdefault("positive_contexts", []) | |
| item.setdefault("negative_contexts", []) | |
| item.setdefault("exceptions", []) | |
| item.setdefault("evidence", []) | |
| item.setdefault("scope", "global") | |
| if item["kind"] not in KIND_VALUES: | |
| item["kind"] = "other" | |
| item["source"] = str(item.get("source", "")).strip() | |
| item["target"] = str(item.get("target", "")).strip() | |
| item["confidence"] = round(clamp(item.get("confidence", 0.0)), 4) | |
| item["evidence_count"] = int(item.get("evidence_count", 0) or 0) | |
| item["positive_evidence_count"] = int(item.get("positive_evidence_count", 0) or 0) | |
| item["negative_evidence_count"] = int(item.get("negative_evidence_count", 0) or 0) | |
| item["explicit_evidence_count"] = int(item.get("explicit_evidence_count", 0) or 0) | |
| if not item["positive_evidence_count"] and item["evidence_count"]: | |
| item["positive_evidence_count"] = item["evidence_count"] | |
| item["aliases"] = list(dict.fromkeys( | |
| str(alias).strip() | |
| for alias in item.get("aliases", []) | |
| if str(alias).strip() | |
| ))[:12] | |
| item["positive_contexts"] = [ | |
| normalize_context_item(x) | |
| for x in item.get("positive_contexts", []) | |
| if isinstance(x, dict) | |
| ][-MAX_CONTEXTS:] | |
| item["negative_contexts"] = [ | |
| normalize_context_item(x) | |
| for x in item.get("negative_contexts", []) | |
| if isinstance(x, dict) | |
| ][-MAX_CONTEXTS:] | |
| item["exceptions"] = [ | |
| normalize_context_item(x) | |
| for x in item.get("exceptions", []) | |
| if isinstance(x, dict) | |
| ][-MAX_EXCEPTIONS:] | |
| item["evidence"] = [ | |
| dict(x) for x in item.get("evidence", []) if isinstance(x, dict) | |
| ][-MAX_EVIDENCE:] | |
| return item | |
| def _contexts_similarity(scene_text: str, note_title: str, contexts: list[dict[str, Any]]) -> float: | |
| if not contexts: | |
| return 0.0 | |
| return max( | |
| context_overlap( | |
| scene_text, | |
| note_title, | |
| item.get("context", ""), | |
| item.get("tags", []), | |
| ) * float(item.get("strength", 1.0)) | |
| for item in contexts | |
| ) | |
| def memory_context_score( | |
| memory: dict[str, Any], | |
| scene: dict[str, Any], | |
| ) -> dict[str, float]: | |
| scene_text = scene.get("text", "") | |
| note_title = scene.get("note_title", "") | |
| positive = _contexts_similarity( | |
| scene_text, | |
| note_title, | |
| memory.get("positive_contexts", []), | |
| ) | |
| negative = _contexts_similarity( | |
| scene_text, | |
| note_title, | |
| memory.get("negative_contexts", []), | |
| ) | |
| if not memory.get("positive_contexts") and memory.get("scope", "global") == "global": | |
| positive = 0.62 | |
| exception_score = _contexts_similarity( | |
| scene_text, | |
| note_title, | |
| memory.get("exceptions", []), | |
| ) | |
| comparison_text = scene.get("input_text", "") or scene_text | |
| for item in memory.get("exceptions", []): | |
| if item.get("note_title") != note_title: | |
| continue | |
| exception_score = max( | |
| exception_score, | |
| similarity(comparison_text, item.get("context", "")), | |
| similarity(normalize_text(comparison_text), normalize_text(item.get("context", ""))), | |
| ) | |
| return { | |
| "positive": round(positive, 4), | |
| "negative": round(negative, 4), | |
| "exception": round(exception_score, 4), | |
| } | |
| def recompute_statuses(memories: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| normalized = [normalize_memory(memory) for memory in (memories or [])] | |
| for memory in normalized: | |
| if memory["status"] == "rejected": | |
| continue | |
| positive = memory["positive_evidence_count"] | |
| negative = memory["negative_evidence_count"] | |
| explicit = memory["explicit_evidence_count"] | |
| if explicit > 0 and memory["confidence"] >= EXPLICIT_ACTIVE_CONFIDENCE: | |
| memory["status"] = "active" | |
| elif ( | |
| positive >= MIN_ACTIVE_POSITIVE | |
| and memory["confidence"] >= ACTIVE_CONFIDENCE | |
| ): | |
| memory["status"] = "active" | |
| else: | |
| memory["status"] = "candidate" | |
| by_source: dict[str, list[dict[str, Any]]] = {} | |
| for memory in normalized: | |
| if memory["status"] == "rejected": | |
| continue | |
| by_source.setdefault(normalize_text(memory["source"]), []).append(memory) | |
| for group in by_source.values(): | |
| targets = {normalize_text(memory["target"]): memory for memory in group} | |
| if len(targets) <= 1: | |
| continue | |
| active = [memory for memory in group if memory["status"] == "active"] | |
| if len(active) <= 1: | |
| continue | |
| for index, left in enumerate(active): | |
| for right in active[index + 1:]: | |
| conflict = 0.0 | |
| for left_ctx in left.get("positive_contexts", []): | |
| for right_ctx in right.get("positive_contexts", []): | |
| conflict = max( | |
| conflict, | |
| context_overlap( | |
| left_ctx.get("context", ""), | |
| left_ctx.get("note_title", ""), | |
| right_ctx.get("context", ""), | |
| right_ctx.get("tags", []), | |
| ), | |
| ) | |
| if conflict >= CONTEXT_CONFLICT_THRESHOLD or ( | |
| not left.get("positive_contexts") | |
| and not right.get("positive_contexts") | |
| ): | |
| left["status"] = "ambiguous" | |
| right["status"] = "ambiguous" | |
| return normalized | |
| def same_memory(memory: dict[str, Any], source: str, target: str) -> bool: | |
| return ( | |
| normalize_text(memory.get("source", "")) == normalize_text(source) | |
| and normalize_text(memory.get("target", "")) == normalize_text(target) | |
| ) | |
| def find_memory(memories: Iterable[dict[str, Any]], source: str, target: str) -> dict[str, Any] | None: | |
| for memory in memories: | |
| if same_memory(memory, source, target): | |
| return memory | |
| return None | |
| def _contains_term(text: str, term: str) -> bool: | |
| if not term: | |
| return False | |
| return bool( | |
| re.search( | |
| rf"(?<!\w){re.escape(term.strip())}(?!\w)", | |
| text or "", | |
| flags=re.IGNORECASE, | |
| ) | |
| ) | |
| def score_word_change(source: str, target: str) -> tuple[float, str]: | |
| source_norm = normalize_text(source) | |
| target_norm = normalize_text(target) | |
| if source_norm == target_norm: | |
| if source == target: | |
| return 0.0, "ignored" | |
| if is_capitalized(target) or has_internal_casing(target): | |
| return 0.78, "casing" | |
| return 0.55, "casing" | |
| if is_stopword(source) or is_stopword(target): | |
| return 0.0, "grammar" | |
| if ( | |
| source_norm in NONLEXICAL_EDIT_TERMS | |
| or target_norm in NONLEXICAL_EDIT_TERMS | |
| ): | |
| return 0.0, "style" | |
| source_tokens = tokenize(source) | |
| target_tokens = tokenize(target) | |
| if source_tokens and target_tokens: | |
| if all(normalize_text(x) in TEMPORAL_TERMS for x in source_tokens + target_tokens): | |
| return 0.0, "temporal" | |
| if all(normalize_text(x) in NONLEXICAL_EDIT_TERMS for x in source_tokens + target_tokens): | |
| return 0.0, "style" | |
| if len(source_tokens) == 1 and len(target_tokens) == 1: | |
| if source_norm in COMMON_STYLE_WORDS or target_norm in COMMON_STYLE_WORDS: | |
| return 0.0, "generic_style" | |
| ratio = similarity(source, target) | |
| score = 0.42 | |
| if ratio >= 0.94: | |
| score += 0.22 | |
| elif ratio >= 0.84: | |
| score += 0.18 | |
| elif ratio >= 0.72: | |
| score += 0.12 | |
| elif ratio >= 0.60: | |
| score += 0.05 | |
| if is_capitalized(target) and not is_capitalized(source): | |
| score += 0.10 | |
| if has_internal_casing(target): | |
| score += 0.08 | |
| if any(char.isdigit() for char in target): | |
| score += 0.08 | |
| specificity = term_specificity(target) | |
| if specificity >= 0.75: | |
| score += 0.12 | |
| elif specificity >= 0.60: | |
| score += 0.08 | |
| if ratio < 0.58 and not is_capitalized(target) and not has_internal_casing(target): | |
| return 0.0, "generic_or_broad" | |
| return round(min(score, 0.96), 4), "lexical_term" | |
| def _compact_alnum(text: str) -> str: | |
| return re.sub(r"[^a-z0-9]", "", (text or "").lower()) | |
| NUMBER_WORDS = { | |
| "zero", "one", "two", "three", "four", "five", "six", "seven", | |
| "eight", "nine", "ten", "eleven", "twelve", "thirteen", "fourteen", | |
| "fifteen", "sixteen", "seventeen", "eighteen", "nineteen", "twenty", | |
| "thirty", "forty", "fifty", "sixty", "seventy", "eighty", "ninety", | |
| } | |
| def _phrase_candidate(old_raw: str, new_raw: str) -> dict[str, Any] | None: | |
| source = re.sub(r"\s+", " ", (old_raw or "").strip()) | |
| target = re.sub(r"\s+", " ", (new_raw or "").strip()) | |
| if not source or not target or source == target: | |
| return None | |
| source_tokens = tokenize(source) | |
| target_tokens = tokenize(target) | |
| if not source_tokens or not target_tokens: | |
| return None | |
| meaningful_source_tokens = [token for token in source_tokens if normalize_text(token) not in {"a", "an", "the"}] | |
| meaningful_target_tokens = [token for token in target_tokens if normalize_text(token) not in {"a", "an", "the"}] | |
| if not meaningful_source_tokens or not meaningful_target_tokens: | |
| return None | |
| if all(token.lower() in COMMON_STYLE_WORDS for token in meaningful_source_tokens + meaningful_target_tokens): | |
| return None | |
| compact_source = _compact_alnum(source) | |
| compact_target = _compact_alnum(target) | |
| if compact_source == compact_target: | |
| punctuation_changed = bool( | |
| re.sub(r"[A-Za-z0-9]", "", source) | |
| != re.sub(r"[A-Za-z0-9]", "", target) | |
| ) | |
| kind = "spelling" if punctuation_changed else "casing" | |
| confidence = 0.82 | |
| else: | |
| compact_ratio = similarity(compact_source, compact_target) | |
| source_token_norms = {normalize_text(token) for token in source_tokens} | |
| acronym_like = any(char.isdigit() for char in target) or len(target.replace(" ", "")) <= 6 and target.replace(" ", "").isalnum() and any(token in NUMBER_WORDS for token in source_token_norms) | |
| if compact_ratio < 0.64 and not acronym_like: | |
| return None | |
| kind = "acronym" if acronym_like else "lexical_term" | |
| confidence = 0.70 if acronym_like and compact_ratio < 0.64 else 0.72 + max(0.0, compact_ratio - 0.72) * 0.45 | |
| return { | |
| "source": source, | |
| "target": target, | |
| "kind": kind, | |
| "confidence": round(min(confidence, 0.93), 4), | |
| } | |
| def candidate_from_groups(old_group: list[str], new_group: list[str]) -> dict[str, Any] | None: | |
| if not old_group or not new_group or len(old_group) > 5 or len(new_group) > 5: | |
| return None | |
| source = " ".join(old_group).strip() | |
| target = " ".join(new_group).strip() | |
| if not source or not target: | |
| return None | |
| old_norm = [normalize_text(x) for x in old_group] | |
| new_norm = [normalize_text(x) for x in new_group] | |
| if old_norm == new_norm: | |
| if source != target: | |
| confidence = 0.80 if is_capitalized(target) or has_internal_casing(target) else 0.74 | |
| return { | |
| "source": source, | |
| "target": target, | |
| "kind": "casing", | |
| "confidence": confidence, | |
| } | |
| return None | |
| if len(old_norm) < len(new_norm): | |
| for start in range(len(new_norm) - len(old_norm) + 1): | |
| if new_norm[start:start + len(old_norm)] == old_norm: | |
| extras = new_norm[:start] + new_norm[start + len(old_norm):] | |
| if extras and all(token in NONLEXICAL_EDIT_TERMS for token in extras): | |
| return None | |
| if len(new_norm) < len(old_norm): | |
| for start in range(len(old_norm) - len(new_norm) + 1): | |
| if old_norm[start:start + len(new_norm)] == new_norm: | |
| extras = old_norm[:start] + old_norm[start + len(new_norm):] | |
| if extras and all(token in NONLEXICAL_EDIT_TERMS for token in extras): | |
| return None | |
| if len(old_group) > 1 or len(new_group) > 1: | |
| phrase = _phrase_candidate(source, target) | |
| if phrase is not None: | |
| return phrase | |
| confidence, kind = score_word_change(source, target) | |
| if confidence <= 0: | |
| return None | |
| compact = similarity(source.replace(" ", ""), target.replace(" ", "")) | |
| if len(old_group) > 1 or len(new_group) > 1: | |
| confidence = max(confidence, 0.62) | |
| if compact >= 0.66: | |
| confidence += 0.07 | |
| if is_capitalized(new_group[0]): | |
| confidence += 0.05 | |
| if compact < 0.46 and not is_capitalized(new_group[0]) and not has_internal_casing(target): | |
| return None | |
| confidence = min(confidence, 0.95) | |
| elif confidence < 0.55: | |
| return None | |
| return { | |
| "source": source, | |
| "target": target, | |
| "kind": kind, | |
| "confidence": round(confidence, 4), | |
| } | |
| def extract_edit_candidates(before: str, after: str) -> list[dict[str, Any]]: | |
| old_tokens = tokenize_with_spans(before) | |
| new_tokens = tokenize_with_spans(after) | |
| old_values = [token["text"] for token in old_tokens] | |
| new_values = [token["text"] for token in new_tokens] | |
| matcher = SequenceMatcher(None, old_values, new_values, autojunk=False) | |
| replace_ops = [opcode for opcode in matcher.get_opcodes() if opcode[0] == "replace"] | |
| if not replace_ops: | |
| return [] | |
| total_old = sum(opcode[2] - opcode[1] for opcode in replace_ops) | |
| total_new = sum(opcode[4] - opcode[3] for opcode in replace_ops) | |
| if total_old > 10 or total_new > 10 or len(replace_ops) > 4: | |
| return [] | |
| candidates: list[dict[str, Any]] = [] | |
| for _, old_start, old_end, new_start, new_end in replace_ops: | |
| old_group = old_values[old_start:old_end] | |
| new_group = new_values[new_start:new_end] | |
| if old_start < len(old_tokens) and old_end > old_start: | |
| old_raw = before[old_tokens[old_start]["start"]:old_tokens[old_end - 1]["end"]] | |
| else: | |
| old_raw = " ".join(old_group) | |
| if new_start < len(new_tokens) and new_end > new_start: | |
| new_raw = after[new_tokens[new_start]["start"]:new_tokens[new_end - 1]["end"]] | |
| else: | |
| new_raw = " ".join(new_group) | |
| if len(old_group) > 1 or len(new_group) > 1: | |
| phrase = _phrase_candidate(old_raw, new_raw) | |
| if phrase is not None: | |
| token_start = new_tokens[new_start]["start"] if new_start < len(new_tokens) else 0 | |
| token_end = new_tokens[new_end - 1]["end"] if new_end > new_start else token_start | |
| phrase["context"] = context_for_change(after, token_start, token_end) | |
| candidates.append(phrase) | |
| continue | |
| expanded = [] | |
| if len(old_group) == len(new_group): | |
| for index, (old_token, new_token) in enumerate(zip(old_group, new_group)): | |
| if normalize_text(old_token) == normalize_text(new_token) and old_token == new_token: | |
| continue | |
| candidate = candidate_from_groups([old_token], [new_token]) | |
| if candidate is not None: | |
| token_start = new_tokens[new_start + index]["start"] | |
| token_end = new_tokens[new_start + index]["end"] | |
| candidate["context"] = context_for_change(after, token_start, token_end) | |
| expanded.append(candidate) | |
| if expanded and len(expanded) <= 3: | |
| candidates.extend(expanded) | |
| continue | |
| candidate = candidate_from_groups(old_group, new_group) | |
| if candidate is None: | |
| continue | |
| if new_start < len(new_tokens) and new_end > new_start: | |
| context_start = new_tokens[new_start]["start"] | |
| context_end = new_tokens[min(new_end - 1, len(new_tokens) - 1)]["end"] | |
| candidate["context"] = context_for_change(after, context_start, context_end) | |
| candidates.append(candidate) | |
| if len(candidates) > 1: | |
| candidates = [candidate for candidate in candidates if candidate["confidence"] >= 0.58] | |
| return candidates[:6] | |
| def context_for_change(text: str, start: int, end: int, radius: int = 140) -> str: | |
| left = max(0, start - radius) | |
| right = min(len(text), end + radius) | |
| return re.sub(r"\s+", " ", text[left:right].strip()) | |
| def _add_positive_context( | |
| memory: dict[str, Any], | |
| note_title: str, | |
| context: str, | |
| strength: float = 1.0, | |
| ) -> None: | |
| memory["positive_contexts"].append( | |
| normalize_context_item( | |
| { | |
| "note_title": note_title, | |
| "context": context, | |
| "tags": extract_context_tags(context, note_title), | |
| "strength": strength, | |
| "timestamp": now_iso(), | |
| } | |
| ) | |
| ) | |
| memory["positive_contexts"] = memory["positive_contexts"][-MAX_CONTEXTS:] | |
| def _add_negative_context( | |
| memory: dict[str, Any], | |
| note_title: str, | |
| context: str, | |
| kind: str, | |
| ) -> None: | |
| memory["negative_contexts"].append( | |
| normalize_context_item( | |
| { | |
| "note_title": note_title, | |
| "context": context, | |
| "tags": extract_context_tags(context, note_title), | |
| "strength": 1.0, | |
| "timestamp": now_iso(), | |
| "type": kind, | |
| } | |
| ) | |
| ) | |
| memory["negative_contexts"] = memory["negative_contexts"][-MAX_CONTEXTS:] | |
| def _exception_exists(memory: dict[str, Any], note_title: str, context: str) -> bool: | |
| return any( | |
| item.get("note_title") == note_title | |
| and context_overlap( | |
| context, | |
| note_title, | |
| item.get("context", ""), | |
| item.get("tags", []), | |
| ) >= 0.76 | |
| for item in memory.get("exceptions", []) | |
| ) | |
| def _record_exception( | |
| memory: dict[str, Any], | |
| note_title: str, | |
| context: str, | |
| reason: str, | |
| ) -> None: | |
| if not _exception_exists(memory, note_title, context): | |
| memory["exceptions"].append( | |
| normalize_context_item( | |
| { | |
| "note_title": note_title, | |
| "context": context, | |
| "tags": extract_context_tags(context, note_title), | |
| "strength": 1.0, | |
| "timestamp": now_iso(), | |
| "type": "one_off_override", | |
| "reason": reason, | |
| } | |
| ) | |
| ) | |
| memory["exceptions"] = memory["exceptions"][-MAX_EXCEPTIONS:] | |
| def detect_generation_overrides( | |
| before: str, | |
| after: str, | |
| memories: list[dict[str, Any]], | |
| last_generation: dict[str, Any] | None, | |
| note_title: str = "", | |
| ) -> tuple[list[dict[str, Any]], dict[str, Any]]: | |
| memories = [normalize_memory(memory) for memory in (memories or [])] | |
| if not last_generation or not isinstance(last_generation, dict): | |
| return memories, {"type": "no_generation_override", "candidates": []} | |
| if last_generation.get("note_title") != note_title: | |
| return memories, {"type": "no_generation_override", "candidates": []} | |
| generation_changes = last_generation.get("changes") or [] | |
| if not generation_changes: | |
| return memories, {"type": "no_generation_override", "candidates": []} | |
| reversal_candidates = extract_edit_candidates(before, after) | |
| if not reversal_candidates: | |
| return memories, {"type": "no_generation_override", "candidates": []} | |
| applied = [] | |
| for change in generation_changes: | |
| memory_id = change.get("memory_id") | |
| original = str(change.get("original", "")) | |
| replacement = str(change.get("replacement", "")) | |
| memory = next( | |
| (item for item in memories if item["id"] == memory_id), | |
| None, | |
| ) | |
| if memory is None: | |
| continue | |
| reversed_here = False | |
| for candidate in reversal_candidates: | |
| candidate_source = str(candidate.get("source", "")) | |
| candidate_target = str(candidate.get("target", "")) | |
| if memory.get("kind") == "casing": | |
| if candidate_source == replacement and candidate_target == original: | |
| reversed_here = True | |
| break | |
| elif ( | |
| normalize_text(candidate_source) == normalize_text(replacement) | |
| and normalize_text(candidate_target) == normalize_text(original) | |
| ): | |
| reversed_here = True | |
| break | |
| if not reversed_here: | |
| continue | |
| context = after | |
| _record_exception( | |
| memory, | |
| note_title, | |
| context, | |
| "The user reversed a Lexi-generated substitution for this scene.", | |
| ) | |
| _add_negative_context( | |
| memory, | |
| note_title, | |
| context, | |
| "generated_override", | |
| ) | |
| memory["negative_evidence_count"] += 1 | |
| memory["evidence_count"] += 1 | |
| memory["updated_at"] = now_iso() | |
| memory["last_reason"] = ( | |
| "The user explicitly overrode this generated substitution for the current scene." | |
| ) | |
| memory.setdefault("evidence", []).append( | |
| { | |
| "type": "generated_override", | |
| "before": before, | |
| "after": after, | |
| "context": context, | |
| "note_title": note_title, | |
| "polarity": "negative", | |
| "timestamp": now_iso(), | |
| } | |
| ) | |
| memory["evidence"] = memory["evidence"][-MAX_EVIDENCE:] | |
| applied.append( | |
| { | |
| "memory_id": memory["id"], | |
| "source": memory["source"], | |
| "target": memory["target"], | |
| } | |
| ) | |
| if not applied: | |
| return memories, {"type": "no_generation_override", "candidates": []} | |
| recompute_statuses(memories) | |
| return memories, { | |
| "type": "memory_override", | |
| "message": ( | |
| "Lexi recorded a contextual override of its own substitution and did not create a reverse memory." | |
| ), | |
| "candidates": applied, | |
| } | |
| def _find_conflicting_memory( | |
| memories: list[dict[str, Any]], | |
| source: str, | |
| target: str, | |
| ) -> dict[str, Any] | None: | |
| source_norm = normalize_text(source) | |
| target_norm = normalize_text(target) | |
| for memory in memories: | |
| memory_source = str(memory.get("source", "")) | |
| memory_target = str(memory.get("target", "")) | |
| if memory.get("kind") in {"casing", "spelling"}: | |
| if ( | |
| memory_source != memory_target | |
| and source == memory_target | |
| and _compact_alnum(target) == _compact_alnum(memory_source) | |
| and target != memory_target | |
| ): | |
| return memory | |
| continue | |
| if ( | |
| source_norm != target_norm | |
| and normalize_text(memory_source) == target_norm | |
| and normalize_text(memory_target) == source_norm | |
| ): | |
| return memory | |
| return None | |
| def detect_memory_reversals( | |
| before: str, | |
| after: str, | |
| memories: list[dict[str, Any]], | |
| note_title: str = "", | |
| ) -> tuple[list[dict[str, Any]], dict[str, Any]]: | |
| """Detect user edits that reverse an existing memory mapping. | |
| This is deterministic and runs before LLM learning. Reversing a known | |
| mapping is negative evidence for that memory in the current context, | |
| not a request to create a reverse memory. | |
| """ | |
| memories = [normalize_memory(memory) for memory in (memories or [])] | |
| candidates = extract_edit_candidates(before, after) | |
| if not candidates: | |
| return memories, { | |
| "type": "no_memory_reversal", | |
| "candidates": [], | |
| } | |
| negative_candidates = [] | |
| seen = set() | |
| for candidate in candidates: | |
| observed_source = str(candidate.get("source", "")).strip() | |
| observed_target = str(candidate.get("target", "")).strip() | |
| if not observed_source or not observed_target: | |
| continue | |
| memory = _find_conflicting_memory( | |
| memories, | |
| observed_source, | |
| observed_target, | |
| ) | |
| if memory is None: | |
| continue | |
| if memory.get("status") == "rejected": | |
| continue | |
| context = str(candidate.get("context", "") or after) | |
| key = ( | |
| memory["id"], | |
| normalize_text(context), | |
| normalize_text(observed_source), | |
| normalize_text(observed_target), | |
| ) | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| _record_exception( | |
| memory, | |
| note_title, | |
| context, | |
| "The user manually reversed a learned memory in this context.", | |
| ) | |
| _add_negative_context( | |
| memory, | |
| note_title, | |
| context, | |
| "manual_reverse", | |
| ) | |
| memory["negative_evidence_count"] += 1 | |
| memory["evidence_count"] += 1 | |
| memory["updated_at"] = now_iso() | |
| memory["last_reason"] = ( | |
| "The user manually reversed this learned substitution in this context." | |
| ) | |
| memory.setdefault("evidence", []).append( | |
| { | |
| "type": "manual_reverse", | |
| "before": before, | |
| "after": after, | |
| "context": context, | |
| "note_title": note_title, | |
| "polarity": "negative", | |
| "timestamp": now_iso(), | |
| } | |
| ) | |
| memory["evidence"] = memory["evidence"][-MAX_EVIDENCE:] | |
| negative_candidates.append( | |
| { | |
| "memory_id": memory["id"], | |
| "source": memory["source"], | |
| "target": memory["target"], | |
| "observed_source": observed_source, | |
| "observed_target": observed_target, | |
| "context": context, | |
| "negative_evidence_count": memory["negative_evidence_count"], | |
| "confidence": round(memory["confidence"], 3), | |
| "action": "negative_evidence", | |
| } | |
| ) | |
| if not negative_candidates: | |
| return memories, { | |
| "type": "no_memory_reversal", | |
| "candidates": [], | |
| } | |
| recompute_statuses(memories) | |
| return memories, { | |
| "type": "memory_negative_evidence", | |
| "message": ( | |
| "Lexi recorded negative evidence for the existing memory " | |
| "in this context. It did not create a reverse memory." | |
| ), | |
| "candidates": negative_candidates, | |
| } | |
| def incorporate_llm_proposals( | |
| memories: list[dict[str, Any]], | |
| proposals: list[dict[str, Any]], | |
| before: str, | |
| after: str, | |
| note_title: str = "", | |
| decision: str = "learn", | |
| ) -> tuple[list[dict[str, Any]], dict[str, Any]]: | |
| memories = [normalize_memory(memory) for memory in (memories or [])] | |
| learned = [] | |
| conflicts = [] | |
| for raw in proposals or []: | |
| source = str(raw.get("source", "")).strip() | |
| target = str(raw.get("target", "")).strip() | |
| kind = str(raw.get("kind", "other")) | |
| if not source or not target or source == target: | |
| continue | |
| if normalize_text(source) == normalize_text(target) and kind != "casing": | |
| continue | |
| if not _contains_term(before, source) or not _contains_term(after, target): | |
| continue | |
| if kind not in KIND_VALUES: | |
| kind = "other" | |
| confidence = clamp(raw.get("confidence", 0.0)) | |
| explicit = bool(raw.get("explicit", False)) | |
| rationale = str(raw.get("rationale", "")).strip() | |
| aliases = [source, *raw.get("aliases", [])] | |
| compact_source = _compact_alnum(source) | |
| if compact_source and compact_source != normalize_text(source): | |
| aliases.append(compact_source) | |
| scope = str(raw.get("scope", "global")) | |
| if scope not in {"global", "note"}: | |
| scope = "global" | |
| reverse = _find_conflicting_memory(memories, source, target) | |
| if reverse is not None: | |
| _add_negative_context(reverse, note_title, after, "reverse_evidence") | |
| reverse["negative_evidence_count"] += 1 | |
| reverse["evidence_count"] += 1 | |
| reverse["updated_at"] = now_iso() | |
| reverse["last_reason"] = rationale or "The user supplied evidence that conflicts with an existing mapping." | |
| conflicts.append( | |
| { | |
| "existing_memory_id": reverse["id"], | |
| "existing_mapping": f'{reverse["source"]} → {reverse["target"]}', | |
| "observed_mapping": f"{source} → {target}", | |
| } | |
| ) | |
| continue | |
| existing = find_memory(memories, source, target) | |
| evidence_context = str(raw.get("context", "") or after) | |
| evidence = { | |
| "type": raw.get("evidence_type", "direct_correction"), | |
| "before": before, | |
| "after": after, | |
| "context": evidence_context, | |
| "note_title": note_title, | |
| "confidence": confidence, | |
| "explicit": explicit, | |
| "timestamp": now_iso(), | |
| } | |
| if existing is None: | |
| existing = normalize_memory( | |
| { | |
| "id": f"mem_{uuid.uuid4().hex[:10]}", | |
| "source": source, | |
| "target": target, | |
| "kind": kind, | |
| "confidence": confidence, | |
| "evidence_count": 1, | |
| "positive_evidence_count": 1, | |
| "negative_evidence_count": 0, | |
| "explicit_evidence_count": 1 if explicit else 0, | |
| "status": "candidate", | |
| "created_at": now_iso(), | |
| "updated_at": now_iso(), | |
| "evidence": [evidence], | |
| "aliases": list(dict.fromkeys(aliases))[:12], | |
| "scope": scope, | |
| "last_reason": rationale, | |
| } | |
| ) | |
| memories.append(existing) | |
| _add_positive_context(existing, note_title, evidence_context, 1.0) | |
| action = "created" | |
| else: | |
| existing["positive_evidence_count"] += 1 | |
| existing["evidence_count"] += 1 | |
| existing["confidence"] = clamp( | |
| max(existing["confidence"], confidence) + (0.06 if explicit else 0.06) | |
| ) | |
| if explicit: | |
| existing["explicit_evidence_count"] += 1 | |
| existing["updated_at"] = now_iso() | |
| existing["last_reason"] = rationale or existing.get("last_reason", "") | |
| existing["aliases"] = list(dict.fromkeys( | |
| existing.get("aliases", []) + aliases | |
| ))[:12] | |
| existing["scope"] = scope if scope == "note" else existing.get("scope", "global") | |
| existing.setdefault("evidence", []).append(evidence) | |
| existing["evidence"] = existing["evidence"][-MAX_EVIDENCE:] | |
| _add_positive_context(existing, note_title, evidence_context, 1.0) | |
| action = "reinforced" | |
| learned.append( | |
| { | |
| "memory_id": existing["id"], | |
| "source": existing["source"], | |
| "target": existing["target"], | |
| "kind": existing["kind"], | |
| "confidence": round(existing["confidence"], 3), | |
| "positive_evidence_count": existing["positive_evidence_count"], | |
| "action": action, | |
| } | |
| ) | |
| memories = recompute_statuses(memories) | |
| if conflicts and not learned: | |
| return memories, { | |
| "type": "memory_conflict", | |
| "message": "Lexi recorded conflicting contextual evidence without creating a reverse mapping.", | |
| "candidates": [], | |
| "conflicts": conflicts, | |
| } | |
| if not learned and decision != "learn": | |
| return memories, { | |
| "type": "ignored_edit", | |
| "message": "Lexi decided this edit was not durable personal lexical evidence.", | |
| "candidates": [], | |
| "conflicts": conflicts, | |
| } | |
| active_learned = [ | |
| item | |
| for item in learned | |
| if next( | |
| (memory for memory in memories if memory["id"] == item["memory_id"]), | |
| {}, | |
| ).get("status") == "active" | |
| ] | |
| event_type = "memory_learned" if active_learned else "memory_candidate" | |
| if conflicts: | |
| event_type = "memory_conflict" | |
| if active_learned: | |
| message = "Lexi learned or reinforced: " + ", ".join( | |
| f'{item["source"]} → {item["target"]}' for item in active_learned | |
| ) + "." | |
| elif conflicts: | |
| message = "Lexi kept the existing memory and recorded conflicting contextual evidence." | |
| else: | |
| message = "Lexi stored the correction as candidate personal vocabulary." | |
| return memories, { | |
| "type": event_type, | |
| "message": message, | |
| "candidates": learned, | |
| "conflicts": conflicts, | |
| } | |
| def learn_from_edit( | |
| before: str, | |
| after: str, | |
| memories: list[dict[str, Any]], | |
| previous_edit: dict[str, Any] | None = None, | |
| last_generation: dict[str, Any] | None = None, | |
| note_title: str = "", | |
| ) -> tuple[list[dict[str, Any]], dict[str, Any]]: | |
| memories = [normalize_memory(memory) for memory in (memories or [])] | |
| if before == after: | |
| return memories, {"type": "no_change", "message": "No manual edit detected.", "candidates": []} | |
| if previous_edit and previous_edit.get("before") == after and previous_edit.get("after") == before: | |
| return memories, { | |
| "type": "undo", | |
| "message": "Lexi ignored an exact undo of the previous manual edit.", | |
| "candidates": [], | |
| } | |
| override_memories, override_event = detect_generation_overrides( | |
| before, | |
| after, | |
| memories, | |
| last_generation, | |
| note_title, | |
| ) | |
| if override_event["type"] == "memory_override": | |
| return override_memories, override_event | |
| negative_memories, negative_event = detect_memory_reversals( | |
| before, | |
| after, | |
| memories, | |
| note_title, | |
| ) | |
| if negative_event["type"] == "memory_negative_evidence": | |
| return negative_memories, negative_event | |
| candidates = extract_edit_candidates(before, after) | |
| if not candidates: | |
| return memories, { | |
| "type": "ignored_edit", | |
| "message": "The edit did not contain a localized lexical correction worth learning.", | |
| "candidates": [], | |
| } | |
| scene = build_scene(note_title, before, after) | |
| proposals = [] | |
| for candidate in candidates: | |
| source = candidate["source"] | |
| target = candidate["target"] | |
| confidence = candidate["confidence"] | |
| if candidate.get("kind") in {"generic_style", "temporal", "grammar", "style"}: | |
| continue | |
| proposals.append( | |
| { | |
| "source": source, | |
| "target": target, | |
| "kind": candidate.get("kind", "other"), | |
| "confidence": confidence, | |
| "rationale": "A localized edit provided evidence of a preferred lexical form.", | |
| "aliases": [source], | |
| "evidence_type": "direct_correction", | |
| "explicit": False, | |
| "scope": "global", | |
| "context": candidate.get("context", scene["text"]), | |
| } | |
| ) | |
| if not proposals: | |
| return memories, { | |
| "type": "ignored_edit", | |
| "message": "Lexi treated this change as ordinary writing rather than personal vocabulary.", | |
| "candidates": candidates, | |
| } | |
| return incorporate_llm_proposals( | |
| memories, | |
| proposals, | |
| before, | |
| after, | |
| note_title=note_title, | |
| ) | |
| def active_memories(memories: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| return [ | |
| memory | |
| for memory in recompute_statuses( | |
| [normalize_memory(memory) for memory in (memories or [])] | |
| ) | |
| if ( | |
| memory["status"] == "active" | |
| and ( | |
| normalize_text(memory["source"]) | |
| != normalize_text(memory["target"]) | |
| or memory.get("kind") == "casing" | |
| ) | |
| ) | |
| ] | |
| def memory_pattern(source: str) -> str: | |
| parts = source.split() | |
| if not parts: | |
| return "" | |
| escaped = [re.escape(part) for part in parts] | |
| body = escaped[0] if len(escaped) == 1 else r"\s+".join(escaped) | |
| return rf"(?<!\w){body}(?!\w)" | |
| def _context_allows_memory(memory: dict[str, Any], scene: dict[str, Any], positive: float, negative: float) -> bool: | |
| if negative >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD and negative > positive: | |
| return False | |
| current_tags = set(scene.get("tags", [])) | |
| stored_tags = { | |
| tag | |
| for item in memory.get("positive_contexts", []) | |
| for tag in item.get("tags", []) | |
| } | |
| kind = memory.get("kind", "other") | |
| if current_tags.intersection(stored_tags): | |
| return True | |
| if not current_tags or not stored_tags: | |
| return True | |
| if current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): | |
| return False | |
| if kind == "person_name" and "physical" in current_tags and not current_tags.intersection({"work", "software", "education"}): | |
| return False | |
| if kind in SPECIFIC_KINDS and current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): | |
| return False | |
| if not stored_tags: | |
| return positive >= CONTEXT_APPLY_THRESHOLD or positive >= 0.60 | |
| return positive >= CONTEXT_APPLY_THRESHOLD | |
| def _span_protection_reason(text: str, start: int, end: int) -> str: | |
| value = text or "" | |
| prefix = value[:start] | |
| suffix = value[end:] | |
| quote_pairs = [ | |
| (prefix.rfind('"'), suffix.find('"')), | |
| (prefix.rfind("'"), suffix.find("'")), | |
| (prefix.rfind("“"), suffix.find("”")), | |
| ] | |
| for opening, closing in quote_pairs: | |
| if opening >= 0 and closing >= 0: | |
| return "inside quoted text" | |
| local = context_for_change(value, start, end, radius=90) | |
| lowered = normalize_text(local) | |
| historical_patterns = [ | |
| r"\bold transcript\b", | |
| r"\bprevious transcript\b", | |
| r"\bprevious draft\b", | |
| r"\bhistorical quote\b", | |
| r"\blegacy spelling\b", | |
| r"\bformerly\b", | |
| r"\bused to (?:call|be|use)\b", | |
| r"\bbefore (?:it|we) (?:renamed|changed)\b", | |
| r"\b(?:old|previous|earlier) (?:name|version|term|spelling|draft|transcript|quote)\b", | |
| r"\bthe old .*?(?:said|says|wrote|read|used|called|spelled)\b", | |
| r"\b(?:old|previous|earlier) (?:design|architecture|technical|product) (?:doc|document|draft)\b.*?\b(?:said|says|wrote|read|used|called|spelled)\b", | |
| ] | |
| if any(re.search(pattern, lowered) for pattern in historical_patterns): | |
| return "historical or legacy reference" | |
| if re.search(r"\b(?:do not|don't|never|avoid)\b", lowered): | |
| return "negated or contrastive reference" | |
| if re.search(r"\bnot\s+(?:the|this|that|a|an)?\s*\w+", lowered): | |
| return "negated or contrastive reference" | |
| if re.search(r"\brather than\b", lowered): | |
| return "negated or contrastive reference" | |
| meta_patterns = [ | |
| r"\bthe word\b", | |
| r"\bspelling\b", | |
| r"\bquoted\b", | |
| r"\bquote\b", | |
| r"\bwhat .*? is called\b", | |
| ] | |
| if any(re.search(pattern, lowered) for pattern in meta_patterns): | |
| return "meta-language reference" | |
| return "" | |
| def apply_memories( | |
| text: str, | |
| memories: list[dict[str, Any]], | |
| selected_ids: set[str] | None = None, | |
| note_title: str = "", | |
| scene_text: str = "", | |
| ) -> tuple[str, list[dict[str, Any]]]: | |
| if not text: | |
| return text, [] | |
| scene = build_scene(note_title, scene_text or "", text) | |
| normalized_memories = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) | |
| candidates = [ | |
| memory for memory in normalized_memories | |
| if memory.get("status") in {"active", "candidate"} | |
| ] | |
| if selected_ids is not None: | |
| candidates = [memory for memory in candidates if memory["id"] in selected_ids] | |
| changes: list[dict[str, Any]] = [] | |
| replacement_pairs = [] | |
| for memory in candidates: | |
| if memory.get("scope") == "note": | |
| allowed_notes = { | |
| item.get("note_title") | |
| for item in memory.get("positive_contexts", []) | |
| if item.get("note_title") | |
| } | |
| if allowed_notes and note_title not in allowed_notes: | |
| continue | |
| score = memory_context_score(memory, scene) | |
| if score["exception"] >= 0.76: | |
| continue | |
| if score["negative"] > score["positive"] and score["negative"] >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD: | |
| continue | |
| current_tags = set(scene.get("tags", [])) | |
| stored_tags = { | |
| tag | |
| for item in memory.get("positive_contexts", []) | |
| for tag in item.get("tags", []) | |
| } | |
| if current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): | |
| continue | |
| if memory.get("kind") in SPECIFIC_KINDS and current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): | |
| continue | |
| effective = max(score["positive"], 0.60 if not memory.get("positive_contexts") else 0.0) | |
| if effective < CONTEXT_APPLY_THRESHOLD and not _context_allows_memory(memory, scene, score["positive"], score["negative"]): | |
| continue | |
| for term in list(dict.fromkeys([memory["source"], *memory.get("aliases", [])])): | |
| if ( | |
| term | |
| and ( | |
| normalize_text(term) != normalize_text(memory["target"]) | |
| or memory.get("kind") == "casing" | |
| ) | |
| ): | |
| replacement_pairs.append((memory, term, score)) | |
| replacement_pairs.sort(key=lambda item: len(item[1]), reverse=True) | |
| result = text | |
| for memory, source, score in replacement_pairs: | |
| pattern = memory_pattern(source) | |
| if not pattern: | |
| continue | |
| regex = re.compile(pattern, re.IGNORECASE) | |
| def replace(match: re.Match[str]) -> str: | |
| original = match.group(0) | |
| if ( | |
| normalize_text(original) == normalize_text(memory["target"]) | |
| and memory.get("kind") != "casing" | |
| ): | |
| return original | |
| if memory.get("kind") == "casing" and original == memory["target"]: | |
| return original | |
| protection = _span_protection_reason(result, match.start(), match.end()) | |
| if protection: | |
| return original | |
| local_context = context_for_change(result, match.start(), match.end()) | |
| local_scene = build_scene(note_title, scene_text or "", local_context) | |
| local_score = memory_context_score(memory, local_scene) | |
| if local_score["exception"] >= 0.76: | |
| return original | |
| if local_score["negative"] > local_score["positive"] and local_score["negative"] >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD: | |
| return original | |
| effective = max(local_score["positive"], 0.60 if not memory.get("positive_contexts") else 0.0) | |
| if effective < CONTEXT_APPLY_THRESHOLD and not _context_allows_memory(memory, local_scene, local_score["positive"], local_score["negative"]): | |
| return original | |
| memory["last_used_at"] = now_iso() | |
| changes.append( | |
| { | |
| "original": original, | |
| "replacement": memory["target"], | |
| "memory_id": memory["id"], | |
| "reason": memory.get("last_reason") or "Contextually relevant personal lexical memory.", | |
| "confidence": round(memory["confidence"], 3), | |
| "context_score": round(effective, 3), | |
| "positive_context_score": round(local_score["positive"], 3), | |
| "negative_context_score": round(local_score["negative"], 3), | |
| "note_title": note_title, | |
| } | |
| ) | |
| return memory["target"] | |
| result = regex.sub(replace, result) | |
| deduped = [] | |
| seen = set() | |
| for change in changes: | |
| key = (change["memory_id"], change["original"], change["replacement"]) | |
| if key not in seen: | |
| seen.add(key) | |
| deduped.append(change) | |
| return result, deduped | |
| def memory_context( | |
| memories: list[dict[str, Any]], | |
| selected_ids: set[str] | None = None, | |
| limit: int = 8, | |
| ) -> list[dict[str, Any]]: | |
| normalized = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) | |
| active = [ | |
| memory for memory in normalized | |
| if memory.get("status") in {"active", "candidate"} | |
| and ( | |
| selected_ids is None | |
| or memory["id"] in selected_ids | |
| ) | |
| ] | |
| active.sort( | |
| key=lambda memory: ( | |
| memory["confidence"], | |
| memory["positive_evidence_count"], | |
| -memory["negative_evidence_count"], | |
| ), | |
| reverse=True, | |
| ) | |
| return [ | |
| { | |
| "id": memory["id"], | |
| "source": memory["source"], | |
| "target": memory["target"], | |
| "kind": memory["kind"], | |
| "confidence": round(memory["confidence"], 3), | |
| "positive_evidence_count": memory["positive_evidence_count"], | |
| "negative_evidence_count": memory["negative_evidence_count"], | |
| "aliases": memory.get("aliases", []), | |
| "scope": memory.get("scope", "global"), | |
| "positive_contexts": memory.get("positive_contexts", [])[-4:], | |
| "negative_contexts": memory.get("negative_contexts", [])[-4:], | |
| "exceptions": memory.get("exceptions", [])[-4:], | |
| } | |
| for memory in active[:limit] | |
| ] | |
| def memory_summary(memories: list[dict[str, Any]]) -> dict[str, Any]: | |
| normalized = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) | |
| normalized.sort( | |
| key=lambda memory: ( | |
| 0 if memory["status"] == "active" else 1 if memory["status"] == "candidate" else 2, | |
| -memory["confidence"], | |
| ) | |
| ) | |
| return { | |
| "memories": normalized, | |
| "active_count": sum(memory["status"] == "active" for memory in normalized), | |
| "candidate_count": sum(memory["status"] == "candidate" for memory in normalized), | |
| "ambiguous_count": sum(memory["status"] == "ambiguous" for memory in normalized), | |
| } | |
| def trim_history(history: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| return list(history or [])[-MAX_HISTORY:] | |