"""PALIMPSESTE — Soft Phi: continuous retrieval with exponential weighting. The standard Phi uses a hard Hamming cutoff: if a candidate's distance exceeds the radius, it's ignored entirely. This is binary — one bit of difference can change the result completely. Soft Phi replaces this with a continuous weighting: ALL candidates contribute, weighted by exp(similarity * temperature). Close candidates dominate, distant ones contribute negligibly. The result is a *blend* of all stored values, not just the nearest one. This is the hypervectorial equivalent of a softmax over the memory. It enables **interpolation**: if the query is between two stored facts, the result blends both rather than picking one arbitrarily. Example ------- If M contains: ("capital of france", "paris") — similarity 0.8 to query ("capital of italy", "rome") — similarity 0.6 to query ("capital of germany", "berlin") — similarity 0.5 to query And the query is "capital of spain" (not stored): - Hard Phi: no match (all similarities < threshold) → fallback - Soft Phi: blend of paris (0.8 weight), rome (0.6), berlin (0.5) → produces an HV in the "capital city" region of hyperspace The blended result won't be exactly "madrid" (that would require storing it), but it will be in the right *semantic neighborhood* — which, combined with HV-Word2Vec, means the model can produce a reasonable guess rather than a blank stare. """ from __future__ import annotations from dataclasses import dataclass, field import numpy as np from .hv import HV, bind, bundle, similarity, hamming, bits_to_signs, signs_to_bits from .memory import Memory from .phi import Phi, KernelConfig, Retrieval __all__ = ["SoftPhi", "SoftPhiConfig"] @dataclass class SoftPhiConfig: """Configuration for Soft Phi. Parameters ---------- temperature : float Sharpness of the weighting. Higher = more peaked (only the closest candidate matters). Lower = more spread (all candidates blend). Typical: 5.0-20.0. At 0.0, all candidates contribute equally. max_candidates : int Maximum number of candidates to consider (for speed). The LSH index returns candidates; we keep only the top-k by similarity. min_weight : float Candidates with decayed memory weight below this are ignored. """ temperature: float = 10.0 max_candidates: int = 100 min_weight: float = 1e-6 @dataclass class SoftPhi: """Soft (continuous) read-back kernel. Unlike :class:`Phi` which uses a hard Hamming cutoff, SoftPhi weights all candidates by exp(similarity * temperature) and produces a blended result. This enables interpolation between stored facts. Parameters ---------- config : SoftPhiConfig Configuration. """ config: SoftPhiConfig = field(default_factory=SoftPhiConfig) def __call__(self, mem: Memory, query: HV) -> HV | None: """Compute Soft Phi(query) — the blended reconstructed value.""" cand_ids = mem.candidates(query) if not cand_ids: return None # filter valid ids and limit to max_candidates valid_ids = [cid for cid in cand_ids if 0 <= cid < len(mem.traces)] if not valid_ids: return None # compute similarities and weights for all candidates D = query.D packed_len = len(query.bits) # vectorized: stack candidate address bits n = min(len(valid_ids), self.config.max_candidates) valid_ids = valid_ids[:n] # compute Hamming distances (vectorized with popcount table) cand_bits = np.empty((n, packed_len), dtype=np.uint8) for i, cid in enumerate(valid_ids): cand_bits[i] = mem.traces[cid].address.bits xored = np.bitwise_xor(cand_bits, query.bits[np.newaxis, :]) _POPCOUNT = np.array([bin(i).count('1') for i in range(256)], dtype=np.uint16) hamming_dists = _POPCOUNT[xored].sum(axis=1) sims = 1.0 - 2.0 * hamming_dists / D # (n,) in [-1, 1] # compute exponential weights T = self.config.temperature # shift for numerical stability sims_shifted = sims * T - (sims.max() * T) exp_weights = np.exp(sims_shifted) exp_weights = exp_weights / exp_weights.sum() # get value HVs and memory weights value_signs = np.zeros((n, D), dtype=np.float64) mem_weights = np.ones(n, dtype=np.float64) for i, cid in enumerate(valid_ids): tr = mem.traces[cid] w = mem.current_weight(tr) if w < self.config.min_weight: exp_weights[i] = 0.0 else: mem_weights[i] = w value_signs[i] = bits_to_signs(tr.value).astype(np.float64) # combined weight = exp_weight * mem_weight combined = exp_weights * mem_weights total = combined.sum() if total <= 0: return None combined = combined / total # weighted sum of value signs → sign → HV weighted_sum = (combined[:, np.newaxis] * value_signs).sum(axis=0) out_signs = np.where(weighted_sum >= 0, np.int8(1), np.int8(-1)) return signs_to_bits(out_signs) def retrieve(self, mem: Memory, query: HV) -> Retrieval: """Return a Retrieval object (for compatibility with Phi).""" cand_ids = mem.candidates(query) valid_ids = [cid for cid in cand_ids if 0 <= cid < len(mem.traces)] if not valid_ids: return Retrieval(query=query, config=KernelConfig(), matches=[], sims=[], weights=[], result=None) D = query.D packed_len = len(query.bits) n = min(len(valid_ids), self.config.max_candidates) valid_ids = valid_ids[:n] cand_bits = np.empty((n, packed_len), dtype=np.uint8) for i, cid in enumerate(valid_ids): cand_bits[i] = mem.traces[cid].address.bits xored = np.bitwise_xor(cand_bits, query.bits[np.newaxis, :]) _POPCOUNT = np.array([bin(i).count('1') for i in range(256)], dtype=np.uint16) hamming_dists = _POPCOUNT[xored].sum(axis=1) sims = (1.0 - 2.0 * hamming_dists / D).tolist() from .memory import Trace matches = [mem.traces[cid] for cid in valid_ids] weights = [mem.current_weight(tr) for tr in matches] return Retrieval(query=query, config=KernelConfig(), matches=matches, sims=sims, weights=weights, result=None)