File size: 6,546 Bytes
1f71c7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """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)
|