"""PALIMPSESTE — Cortex layer: expertise, dreaming, compositional reasoning. Three features that make PALIMPSESTE fundamentally different from any LLM: 1. **InstantExpert**: ingest any text document and become an expert instantly. No training, no epoch, no gradient. O(N) one-pass write. Every fact in the document becomes immediately retrievable. 2. **Dreamer**: when idle, the system replays its memory, discovers co- activations, and creates new abstract concepts — becoming smarter without any new data. This is sleep consolidation for an associative memory. 3. **Composer**: decomposes complex questions into sub-questions, resolves each via multi-hop chaining, and composes a final answer. This is compositional reasoning over the HV substrate. No transformer. No attention matrix. No gradient. No GPU. """ from __future__ import annotations import re import time from dataclasses import dataclass, field from typing import Optional import numpy as np from .hv import HV, bind, bundle, similarity, random_hv from .consolidation import Consolidator, ConsolidationConfig, ConsolidationResult from .abstraction import AbstractionEngine, AbstractionConfig, Concept __all__ = [ "InstantExpert", "Dreamer", "Composer", "ExpertiseResult", "DreamResult", "CompositionResult", "MultimodalFusion", "ModalityBinding", "MetaLearner", "MetaLearningResult", "DualMemory", "MemoryRecord", "Analogizer", "AnalogyResult", ] # ================================================================ # 1. INSTANT EXPERTISE # ================================================================ @dataclass class ExpertiseResult: """Result of learning from a text document.""" n_tokens: int n_facts: int n_seconds: float facts: list[tuple[str, str]] document_tag: str class InstantExpert: """Ingest text documents and become an expert instantly. Extracts factual statements from text, writes them to memory as Q/A episodes, and registers them for fuzzy matching. After ingestion, the model can answer questions about the document content — all in O(N) time, no gradient, no epoch. A transformer would need fine-tuning to learn from a document. PALIMPSESTE learns in one pass. """ def __init__(self, lm) -> None: self.lm = lm self._documents: dict[str, list[str]] = {} # tag -> sentences def learn_from_text( self, text: str, document_tag: str | None = None, verbose: bool = False, ) -> ExpertiseResult: """Ingest a text document and become an expert on its content. Parameters ---------- text : str The document text (any length). document_tag : str | None A label for the document (auto-generated if None). verbose : bool Print progress. Returns ------- ExpertiseResult """ t0 = time.perf_counter() if document_tag is None: document_tag = f"doc{len(self._documents)}" # Step 1: extract sentences as potential Q/A pairs facts = self._extract_facts(text) n_facts = len(facts) # Step 2: also learn the raw text for token-level retrieval n_tokens = self.lm.train_on_text(text, verbose=False) # Step 3: if we have a Conversation layer, teach it the Q/A pairs if hasattr(self.lm, '_conversation') and self.lm._conversation: conv = self.lm._conversation for q, a in facts: conv.teach(q, a) # Store self._documents[document_tag] = [a for _, a in facts] dt = time.perf_counter() - t0 if verbose: print(f" Learned {n_tokens} tokens, {n_facts} facts in {dt:.1f}s") return ExpertiseResult( n_tokens=n_tokens, n_facts=n_facts, n_seconds=dt, facts=facts, document_tag=document_tag, ) def _extract_facts(self, text: str) -> list[tuple[str, str]]: """Extract factual Q/A pairs from text. Strategy: split into sentences. For each declarative sentence, create a "what is X" or "tell me about X" question. For sentences with "is/are/was/were", create definition questions. """ facts: list[tuple[str, str]] = [] sentences = self._split_sentences(text) for sent in sentences: sent = sent.strip() if len(sent) < 10 or len(sent) > 200: continue # Pattern: "X is/are/was/were Y" -> Q: "what is X", A: full sentence m = re.match( r'^(.{3,60}?)\s+(?:is|are|was|were)\s+(.+)$', sent, re.IGNORECASE ) if m: subject = m.group(1).strip().rstrip(',').lower() # Create a question from the subject q = f"what is {subject}" facts.append((q, sent)) continue # Pattern: "X defined as Y" / "X means Y" m = re.match( r'^(.{3,60}?)\s+(?:defined as|means|refers to)\s+(.+)$', sent, re.IGNORECASE ) if m: subject = m.group(1).strip().lower() q = f"what is {subject}" facts.append((q, sent)) continue # Fallback: generic "tell me about" for longer sentences if len(sent) > 20: # Extract the first few words as a topic words = sent.split()[:3] topic = ' '.join(words).lower().rstrip(',.;:') q = f"tell me about {topic}" facts.append((q, sent)) return facts[:200] # cap to avoid memory bloat def _split_sentences(self, text: str) -> list[str]: """Split text into sentences.""" # Normalize whitespace text = re.sub(r'\s+', ' ', text) # Split on sentence boundaries parts = re.split(r'(?<=[.!?])\s+', text) return [p.strip() for p in parts if p.strip()] @property def n_documents(self) -> int: return len(self._documents) def documents(self) -> dict[str, list[str]]: return dict(self._documents) # ================================================================ # 2. DREAM CONSOLIDATION # ================================================================ @dataclass class DreamResult: """Result of a dream consolidation cycle.""" n_concepts_promoted: int n_concepts_extracted: int n_cycles: int n_seconds: float concept_labels: list[str] new_connections: list[str] class Dreamer: """Sleep consolidation: replay memory, discover concepts, get smarter. When the system is idle, it "dreams": it replays its memory traces, finds co-activations (things that fire together), and promotes frequently co-active pairs into abstract concepts. It also runs HV clustering to extract concept centroids. A transformer CANNOT do this — its weights are frozen after training. PALIMPSESTE reorganizes its knowledge autonomously. """ def __init__( self, mem, phi=None, rng: np.random.Generator | None = None, config: ConsolidationConfig | None = None, abstraction_config: AbstractionConfig | None = None, ) -> None: self.mem = mem self.rng = rng or np.random.default_rng() self.consolidator = Consolidator( mem=mem, config=config or ConsolidationConfig(), rng=self.rng, ) self.abstraction = AbstractionEngine( mem=mem, config=abstraction_config or AbstractionConfig(), rng=self.rng, ) def dream( self, n_cycles: int = 3, replay_batch: int = 200, verbose: bool = False, ) -> DreamResult: """Run a dream consolidation cycle. 1. Replay: sample random batches of traces and record co-activations. 2. Consolidate: promote frequently co-active pairs to concepts. 3. Abstract: cluster traces into concept centroids. Parameters ---------- n_cycles : int Number of replay-consolidate cycles. replay_batch : int Traces to sample per replay. verbose : bool Print progress. Returns ------- DreamResult """ t0 = time.perf_counter() total_promoted = 0 all_labels: list[str] = [] all_connections: list[str] = [] n_traces = len(self.mem) if n_traces < 10: return DreamResult(0, 0, 0, 0.0, [], []) for cycle in range(n_cycles): if verbose: print(f" Dream cycle {cycle+1}/{n_cycles}...", flush=True) # Phase 1: Replay — sample trace IDs and record co-activations n_sample = min(replay_batch, n_traces) sampled = self.rng.choice(n_traces, size=n_sample, replace=False) # Group co-activations: nearby traces (by ID) are likely related # (they were written during the same episode) for i in range(0, len(sampled), 10): batch = sampled[i:i+10] self.consolidator.observe_retrieval(batch.tolist()) # Phase 2: Consolidate — promote concepts result = self.consolidator.consolidate() total_promoted += len(result.promoted) if verbose and result.promoted: print(f" promoted {len(result.promoted)} concepts", flush=True) # Phase 3: Abstract — cluster extraction concepts = self.abstraction.extract_concepts(verbose=verbose) for c in concepts: all_labels.append(c.label) all_connections.append( f"concept '{c.label}' ({c.n_members} members)" ) dt = time.perf_counter() - t0 return DreamResult( n_concepts_promoted=total_promoted, n_concepts_extracted=len(concepts), n_cycles=n_cycles, n_seconds=dt, concept_labels=all_labels, new_connections=all_connections, ) @property def n_concepts(self) -> int: """Total concepts discovered.""" return self.consolidator.n_promoted + self.abstraction.n_concepts # ================================================================ # 3. COMPOSITIONAL REASONING # ================================================================ @dataclass class CompositionStep: """One step in a compositional reasoning chain.""" step_type: str # "decompose" | "resolve" | "compose" sub_question: str sub_answer: str confidence: float @dataclass class CompositionResult: """Result of compositional reasoning.""" answer: str success: bool steps: list[CompositionStep] n_decompositions: int n_hops: int n_seconds: float # Filler phrases to strip when decomposing questions _FILLER_PATTERNS = [ r'(?:the|a|an)\s+(?:country|city|place|person|thing|element|concept)\s+(?:that|which|who)\s+', r'(?:that|which)\s+(?:won|is|was|has|had|did|do|does)\s+', r'(?:who|what|where|when|why|how)\s+', ] _FILLER_REGEX = [re.compile(p, re.IGNORECASE) for p in _FILLER_PATTERNS] # Decomposition cues — keywords that signal a compositional question _DECOMP_CUES = [ 'that won', 'that is', 'that was', 'that has', 'that had', 'of the country', 'of the city', 'of the person', 'capital of the', 'author of the', 'inventor of the', 'who created', 'who discovered', 'who wrote', 'compared to', 'difference between', ] class Composer: """Compositional reasoning: decompose, resolve, compose. Takes a complex question, decomposes it into simpler sub-questions, resolves each using the associative memory, and composes a final answer. Supports multi-hop chains (A->B->C->D). A transformer does "chain of thought" in its limited context window (temporary, lost after the response). PALIMPSESTE stores each reasoning step as a permanent memory trace — reusable forever. """ def __init__(self, reasoner, verbose: bool = False) -> None: """ Parameters ---------- reasoner : Reasoner The fact-chaining reasoner to use. verbose : bool Print reasoning steps. """ self.reasoner = reasoner self.verbose = verbose def reason( self, question: str, max_depth: int = 3, ) -> CompositionResult: """Answer a complex question via compositional reasoning. Parameters ---------- question : str The complex question. max_depth : int Maximum decomposition depth. Returns ------- CompositionResult """ t0 = time.perf_counter() steps: list[CompositionStep] = [] # Step 1: Check if the question needs decomposition needs_decomp = self._needs_decomposition(question) if not needs_decomp: # Simple question — try direct + chain answer, chain = self.reasoner.respond(question) if answer and not self._is_fallback(answer): steps.append(CompositionStep( step_type="resolve", sub_question=question, sub_answer=answer, confidence=1.0, )) return CompositionResult( answer=answer, success=True, steps=steps, n_decompositions=0, n_hops=0, n_seconds=time.perf_counter() - t0, ) # Step 2: Decompose the question sub_questions = self._decompose(question) if self.verbose: print(f" Decomposed into {len(sub_questions)} sub-questions", flush=True) if not sub_questions: # Can't decompose — try direct chaining answer, chain = self.reasoner.respond(question) if chain and chain.success: for s in chain.steps: steps.append(CompositionStep( step_type="resolve", sub_question=s.sub_question, sub_answer=s.sub_answer, confidence=0.7, )) return CompositionResult( answer=chain.answer, success=True, steps=steps, n_decompositions=0, n_hops=chain.n_hops, n_seconds=time.perf_counter() - t0, ) return CompositionResult( answer=answer, success=False, steps=steps, n_decompositions=0, n_hops=0, n_seconds=time.perf_counter() - t0, ) # Step 3: Resolve each sub-question resolved_parts: list[str] = [] for sq in sub_questions: if self.verbose: print(f" Resolving: {sq}", flush=True) steps.append(CompositionStep( step_type="decompose", sub_question=sq, sub_answer="", confidence=0.0, )) answer, chain = self.reasoner.respond(sq) if answer and not self._is_fallback(answer): steps.append(CompositionStep( step_type="resolve", sub_question=sq, sub_answer=answer, confidence=1.0, )) resolved_parts.append(answer) else: # Try chaining if chain and chain.success: steps.append(CompositionStep( step_type="resolve", sub_question=sq, sub_answer=chain.answer, confidence=0.7, )) resolved_parts.append(chain.answer) # Step 4: Compose — re-query with resolved parts if resolved_parts: # The last resolved part is usually the final answer final_answer = resolved_parts[-1] # If there are multiple parts, try to compose if len(resolved_parts) > 1: composed = self._compose(question, resolved_parts) if composed: final_answer = composed steps.append(CompositionStep( step_type="compose", sub_question=question, sub_answer=final_answer, confidence=0.8, )) return CompositionResult( answer=final_answer, success=True, steps=steps, n_decompositions=len(sub_questions), n_hops=len([s for s in steps if s.step_type == "resolve"]), n_seconds=time.perf_counter() - t0, ) return CompositionResult( answer="", success=False, steps=steps, n_decompositions=len(sub_questions), n_hops=0, n_seconds=time.perf_counter() - t0, ) def _needs_decomposition(self, question: str) -> bool: """Check if a question is complex enough to decompose.""" q_lower = question.lower() # Check for compositional cues for cue in _DECOMP_CUES: if cue in q_lower: return True # Check length — long questions often need decomposition if len(question.split()) > 8: return True return False def _decompose(self, question: str) -> list[str]: """Decompose a complex question into sub-questions.""" q = question.lower().strip().rstrip('?') sub_questions: list[str] = [] # Pattern: "what is the X of the Y that Z" # Decompose into: "what Y that Z" then "what is the X of {answer}" m = re.match( r'what\s+(?:is|are)\s+(?:the\s+)?(.+?)\s+of\s+(?:the\s+)?(.+?)\s+(?:that|which|who)\s+(.+)', q ) if m: relation = m.group(1).strip() subject_phrase = m.group(2).strip() condition = m.group(3).strip() sub_questions.append(f"{subject_phrase} that {condition}") sub_questions.append(f"what is the {relation} of {{answer}}") return sub_questions # Pattern: "who X that Y" -> "who X" + check if Y references something m = re.match(r'(.+?)\s+that\s+(.+)', q) if m: first = m.group(1).strip() second = m.group(2).strip() # Try to find a known question in the first part known = self.reasoner.conv._known_questions for kq in known: kq_lower = kq.lower() if kq_lower in first or first in kq_lower: sub_questions.append(kq) # Replace the known part with a placeholder remainder = first.replace(kq_lower, '').strip() if remainder: sub_questions.append(f"{remainder} that {second}") return sub_questions # Fallback: treat first part as a sub-question sub_questions.append(first) return sub_questions # Pattern: comparison "compare X and Y" m = re.match(r'(?:compare|comparison|difference between)\s+(.+?)\s+and\s+(.+)', q) if m: x = m.group(1).strip() y = m.group(2).strip() sub_questions.append(f"what is {x}") sub_questions.append(f"what is {y}") return sub_questions return sub_questions def _compose(self, question: str, parts: list[str]) -> str: """Compose multiple resolved parts into a final answer.""" if len(parts) == 1: return parts[0] # For comparisons, join the parts q_lower = question.lower() if 'compare' in q_lower or 'difference' in q_lower: return f"{parts[0]} vs {parts[1]}" # For chains, the last resolved answer is usually the final return parts[-1] def _is_fallback(self, text: str) -> bool: """Check if the text is a fallback response.""" from .chat import FALLBACK_RESPONSE return text == FALLBACK_RESPONSE or not text.strip() # ================================================================ # 4. MULTI-MODAL FUSION # ================================================================ @dataclass class ModalityBinding: """A stored cross-modal association.""" text: str image_hv: HV text_hv: HV bound_hv: HV class MultimodalFusion: """Cross-modal association in hypervector space. In HV space, everything is a hypervector. Text, images, audio — all can be bound together into the same memory. No separate architecture like CLIP. No separate encoder per modality. learn_image(description, image) → the image HV and text HV are bound and stored. Later, "show me a cat" retrieves the image HV via text→bound association. This is impossible for a text-only LLM. PALIMPSESTE fuses modalities natively because binding is modality-agnostic. """ def __init__(self, mem, encoder, image_encoder=None) -> None: self.mem = mem self.encoder = encoder if image_encoder is None: from .vision import ImageEncoder image_encoder = ImageEncoder(D=encoder.D) self.image_encoder = image_encoder self._bindings: list[ModalityBinding] = [] def learn_image( self, description: str, image: np.ndarray, verbose: bool = False, ) -> ModalityBinding: """Associate an image with a text description. Parameters ---------- description : str Text label/description for the image. image : np.ndarray Image array (H, W, 3) uint8. Returns ------- ModalityBinding """ # Encode both modalities to HV space image_hv = self.image_encoder.encode(image) text_hv = self.encoder.encode_str(description) # Bind them together — this is the cross-modal association bound_hv = bind(image_hv, text_hv) # Store in memory: both the bound HV and individual HVs self.mem.write(bound_hv, image_hv, weight=1.0, tag=f"image:{description}") self.mem.write(text_hv, image_hv, weight=1.0, tag=f"text_to_image:{description}") binding = ModalityBinding( text=description, image_hv=image_hv, text_hv=text_hv, bound_hv=bound_hv, ) self._bindings.append(binding) if verbose: print(f" Bound image '{description}' to memory ({len(self._bindings)} total)") return binding def find_image(self, query: str, top_k: int = 3) -> list[tuple[str, float]]: """Find images matching a text query. Returns list of (description, similarity) pairs. """ query_hv = self.encoder.encode_str(query) results: list[tuple[str, float]] = [] for b in self._bindings: sim = similarity(query_hv, b.text_hv) results.append((b.text, sim)) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k] def describe_image(self, image: np.ndarray) -> list[tuple[str, float]]: """Find text descriptions matching an image. Returns list of (description, similarity) pairs. """ query_hv = self.image_encoder.encode(image) results: list[tuple[str, float]] = [] for b in self._bindings: sim = similarity(query_hv, b.image_hv) results.append((b.text, sim)) results.sort(key=lambda x: x[1], reverse=True) return results @property def n_bindings(self) -> int: return len(self._bindings) # ================================================================ # 5. META-LEARNING # ================================================================ @dataclass class MetaLearningResult: """Result of a meta-learning step.""" accepted: bool param_changed: str old_value: float new_value: float energy_before: float energy_after: float rationale: str class MetaLearner: """Self-tuning kernel parameters under Lyapunov constraint. The system adjusts its own retrieval parameters (kernel_radius, min_weight, temperature) based on domain. It learns that math questions need a tight radius, creative questions need a wide one. This is Axiome 5 in action: meta-parameters in H_meta can be rewritten only if ΔE[surprise] ≤ 0. The acceptance criterion is immutable, bounding recursion by construction. A transformer's hyperparameters are fixed after training. PALIMPSESTE tunes itself at runtime. """ def __init__(self, mem, phi, rng=None) -> None: from .meta import MetaController, LyapunovEnergy, max_radius_invariant self.mem = mem self.phi = phi self.rng = rng or np.random.default_rng() # Set up Lyapunov-bounded meta controller energy = LyapunovEnergy(invariants=[ max_radius_invariant(max_r=mem.D // 2), ]) self.controller = MetaController( mem=mem, phi=phi, energy=energy, rng=self.rng, ) self._domain_profiles: dict[str, dict] = {} def adapt(self, domain: str | None = None) -> MetaLearningResult: """Attempt to improve retrieval parameters. Parameters ---------- domain : str | None Optional domain tag (e.g. "math", "creative") for profiling. Returns ------- MetaLearningResult """ replay = self.controller.build_replay(n=64) decision = self.controller.step(replay, max_proposals=8) if decision and decision.accepted: old_val = decision.energy_before new_val = decision.energy_after param = "kernel_params" result = MetaLearningResult( accepted=True, param_changed=param, old_value=old_val, new_value=new_val, energy_before=decision.energy_before, energy_after=decision.energy_after, rationale=decision.proposal.rationale, ) # Record domain profile if domain: self._domain_profiles[domain] = { "radius": self.controller.config.radius, "min_weight": self.controller.config.min_weight, } return result return MetaLearningResult( accepted=False, param_changed="none", old_value=decision.energy_before if decision else 0, new_value=decision.energy_after if decision else 0, energy_before=decision.energy_before if decision else 0, energy_after=decision.energy_after if decision else 0, rationale="no improvement found", ) def get_profile(self, domain: str) -> dict | None: """Get the learned parameter profile for a domain.""" return self._domain_profiles.get(domain) @property def current_config(self) -> dict: """Current retrieval parameters.""" cfg = self.controller.config return { "radius": cfg.radius, "min_weight": cfg.min_weight, "sharpness": cfg.sharpness, "topk": cfg.topk, } @property def n_adaptations(self) -> int: """Total successful adaptations.""" return len(self.controller.history) # ================================================================ # 6. EPISODIC + SEMANTIC MEMORY # ================================================================ @dataclass class MemoryRecord: """A stored memory with type classification.""" content: str memory_type: str # "episodic" | "semantic" timestamp: float weight: float tag: str class DualMemory: """Episodic + Semantic dual memory system. Humans have two memory systems: - Episodic: "I talked about X with the user at 3pm" (decays over time) - Semantic: "The capital of France is Paris" (persists) This class manages both in the HV substrate. Episodic memories have a short half-life (configurable), semantic memories persist forever. The system can distinguish "I remember you told me X" from "The factual answer is Y". A transformer has no episodic memory at all — it forgets everything after the context window closes. """ def __init__( self, mem, encoder, episodic_half_life: float = 3600.0, # 1 hour ) -> None: self.mem = mem self.encoder = encoder self._records: list[MemoryRecord] = [] self._episodic_half_life = episodic_half_life def store_episodic(self, content: str, tag: str = "") -> MemoryRecord: """Store an episodic memory (conversation event).""" import time as _time hv = self.encoder.encode_str(content) self.mem.write(hv, hv, weight=0.5, tag=f"episodic:{tag}") record = MemoryRecord( content=content, memory_type="episodic", timestamp=_time.time(), weight=0.5, tag=tag, ) self._records.append(record) return record def store_semantic(self, content: str, tag: str = "") -> MemoryRecord: """Store a semantic memory (persistent fact).""" import time as _time hv = self.encoder.encode_str(content) self.mem.write(hv, hv, weight=1.0, tag=f"semantic:{tag}") record = MemoryRecord( content=content, memory_type="semantic", timestamp=_time.time(), weight=1.0, tag=tag, ) self._records.append(record) return record def recall(self, query: str, top_k: int = 5) -> list[tuple[MemoryRecord, float]]: """Recall memories matching the query, with type information.""" query_hv = self.encoder.encode_str(query) scored: list[tuple[MemoryRecord, float]] = [] for record in self._records: record_hv = self.encoder.encode_str(record.content) sim = similarity(query_hv, record_hv) # Episodic memories decay if record.memory_type == "episodic": import time as _time age = _time.time() - record.timestamp decay = 2 ** (-age / self._episodic_half_life) sim *= decay scored.append((record, sim)) scored.sort(key=lambda x: x[1], reverse=True) return scored[:top_k] def forget_old_episodic(self, max_age: float = 7200.0) -> int: """Remove episodic memories older than max_age seconds. This doesn't delete from memory M (append-only), but removes from the recall index so they're no longer retrieved. """ import time as _time now = _time.time() before = len(self._records) self._records = [ r for r in self._records if r.memory_type != "episodic" or (now - r.timestamp) < max_age ] return before - len(self._records) @property def n_episodic(self) -> int: return sum(1 for r in self._records if r.memory_type == "episodic") @property def n_semantic(self) -> int: return sum(1 for r in self._records if r.memory_type == "semantic") @property def total(self) -> int: return len(self._records) # ================================================================ # 7. ANALOGICAL REASONING # ================================================================ @dataclass class AnalogyResult: """Result of an analogical reasoning query.""" a: str b: str c: str answer: str confidence: float similarity: float class Analogizer: """HV algebra for analogical reasoning. Solves "a is to b as c is to ?" using HV algebra: answer_hv = bundle([unbind(bind(a_hv, b_hv), a_hv), c_hv]) Then finds the closest known word/concept. Example: "paris is to france as tokyo is to ?" → computes the relation vector (paris→france) → applies it to tokyo → finds "japan" as the nearest match This is the classic Plate (1995) HRR algebra, applied for real reasoning on the HV substrate. No transformer can do this — they don't have explicit HV algebra operations. """ def __init__(self, word2vec=None, mem=None, encoder=None) -> None: """ Parameters ---------- word2vec : HVWord2Vec | None Trained word embeddings for word-level analogies. mem : Memory | None Memory for HV-level analogies. encoder : Encoder | None Encoder for string→HV conversion. """ self.w2v = word2vec self.mem = mem self.encoder = encoder def analogy( self, a: str, b: str, c: str, top_k: int = 5, ) -> AnalogyResult | None: """Solve "a is to b as c is to ?". Computes the relation vector from a→b, applies it to c, and finds the closest word. Parameters ---------- a, b, c : str "a is to b as c is to ?" top_k : int Number of candidates to return. Returns ------- AnalogyResult | None """ if self.w2v is None: return None a_hv = self.w2v.get_word_hv(a) b_hv = self.w2v.get_word_hv(b) c_hv = self.w2v.get_word_hv(c) if a_hv is None or b_hv is None or c_hv is None: return None # Compute the relation: unbind a from b to get the "a→b" transform # In HV space: relation = bind(b, a) (XOR is self-inverse) relation_hv = bind(b_hv, a_hv) # Apply relation to c: answer_hv = bind(relation, c) answer_hv = bind(relation_hv, c_hv) # Find closest words candidates = self.w2v.most_similar_hv(answer_hv, top_k=top_k) if not candidates: return None best_word, best_sim = candidates[0] confidence = (best_sim + 1.0) / 2.0 # map [-1,1] → [0,1] return AnalogyResult( a=a, b=b, c=c, answer=best_word, confidence=confidence, similarity=best_sim, ) def find_relations(self, word: str, top_k: int = 5) -> list[tuple[str, float]]: """Find words most related to a given word. Returns list of (word, similarity) pairs. """ if self.w2v is None: return [] return self.w2v.most_similar(word, top_k=top_k)