File size: 8,432 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 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | """PALIMPSESTE — The read-back kernel ``Phi`` (Axiomes 2, 4).
Axiome 2 — Le passage avant est une recuperation associative
``Phi(q) = bundle_{(a,v,w) in N_r(q)} (w * v)``
where ``N_r(q)`` is the Hamming neighborhood of radius ``r`` of ``q``.
Retrieval is ``O(log |M|)`` via LSH (independent of D in the *compute*,
since bundling is bitwise).
Axiome 4 — Les parametres sont reconstruits, non stockes
``W_t = Phi(bind(x, s_t))`` — the "weight" exists only for the duration of
one computation. It emerges from memory. So "modifying the parameters" =
"writing better traces into M".
This module implements ``Phi`` as a *reconstruction operator* over a
:class:`Memory`. It is:
- stateless w.r.t. the memory contents (it reads, never writes);
- parameterized by a small, learnable ``KernelConfig`` (radius, threshold
sharpness, min-weight cutoff) whose values are read from ``H_meta`` by
the :class:`MetaController` (``meta.py``);
- sub-linear in ``|M|`` because it operates only on the LSH candidate set.
The kernel also exposes the *raw* retrieved traces and their (similarity,
weight) scores, so callers (curiosity, consolidation, meta) can introspect
*why* a reconstruction came out the way it did — transparency is a design goal.
"""
from __future__ import annotations
from dataclasses import dataclass, field
import math
import numpy as np
from .hv import HV, bundle, hamming, similarity
from .memory import Memory, Trace
__all__ = ["KernelConfig", "Phi", "Retrieval"]
@dataclass
class KernelConfig:
"""The learnable knobs of the read-back kernel ``Phi``.
These are the meta-parameters that live in ``H_meta`` and that the
:class:`MetaController` may rewrite under a Lyapunov constraint.
Attributes
----------
radius : int
Hamming radius defining the neighborhood ``N_r(q)``.
min_weight : float
Traces with decayed weight below this are ignored (soft floor for
"dormant" recall; keeps rare re-activations possible without flooding
recall with noise).
sharpness : float
Temperature for the optional similarity-weighted softmax over
retrieved values. ``sharpness <= 0`` means plain weighted-bundle
(hard majority); ``sharpness > 0`` interpolates values by
``exp(sharpness * sim)``.
topk : int | None
If set, only the ``topk`` most similar candidates contribute. Caps
per-query compute and noise.
"""
radius: int = 50
min_weight: float = 1e-3
sharpness: float = 0.0
topk: int | None = None
def __post_init__(self) -> None:
if self.radius < 0:
raise ValueError("radius must be >= 0")
if self.min_weight <= 0:
raise ValueError("min_weight must be > 0")
if self.topk is not None and self.topk <= 0:
raise ValueError("topk must be positive or None")
def encode(self) -> dict:
"""Serialize for storage in ``H_meta``."""
return {
"radius": self.radius,
"min_weight": self.min_weight,
"sharpness": self.sharpness,
"topk": self.topk,
}
@classmethod
def decode(cls, d: dict) -> "KernelConfig":
return cls(**d)
@dataclass
class Retrieval:
"""Full result of a ``Phi`` call, for introspection."""
query: HV
config: KernelConfig
matches: list[Trace]
sims: list[float]
weights: list[float]
result: HV | None # None if no matches
@property
def n_matches(self) -> int:
return len(self.matches)
@dataclass
class Phi:
"""The associative read-back kernel.
``Phi`` is a *pure reader* of :class:`Memory`. All learning is done by
writing traces into ``M`` (``learner.py``); ``Phi`` only reconstructs.
"""
config: KernelConfig = field(default_factory=KernelConfig)
# ----------------------------------------------------------------- core
def retrieve(self, mem: Memory, query: HV) -> Retrieval:
"""Find ``N_r(query)`` in ``M`` and return matches + scores.
Two-stage: LSH candidates, then exact Hamming filter (vectorized),
then optional top-k and min-weight cutoffs.
The Hamming distance computation is **vectorized**: all candidate
addresses are XOR'd against the query in a single numpy batch, then
popcount'd. This is 10-50x faster than the per-trace Python loop at
scale (200K+ traces).
"""
cand_ids = mem.candidates(query)
if not cand_ids:
return Retrieval(query=query, config=self.config,
matches=[], sims=[], weights=[], result=None)
# filter valid ids
valid_ids = [cid for cid in cand_ids if 0 <= cid < len(mem.traces)]
if not valid_ids:
return Retrieval(query=query, config=self.config,
matches=[], sims=[], weights=[], result=None)
D = query.D
packed_len = len(query.bits)
# --- vectorized Hamming distance ---
# Fast popcount via a 256-entry lookup table (no np.unpackbits needed).
# This is 8x faster than unpack+sum for large candidate sets.
cand_bits = np.empty((len(valid_ids), 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 via lookup table
_POPCOUNT_TABLE = np.array(
[bin(i).count('1') for i in range(256)], dtype=np.uint16)
hamming_dists = _POPCOUNT_TABLE[xored].sum(axis=1)
# filter by radius
within = hamming_dists <= self.config.radius
if not within.any():
return Retrieval(query=query, config=self.config,
matches=[], sims=[], weights=[], result=None)
# compute similarities and weights for survivors
matched_idx = np.where(within)[0]
all_sims = (1.0 - 2.0 * hamming_dists[matched_idx] / D)
now = None
weights: list[float] = []
sims: list[float] = []
matches: list[Trace] = []
for j, mi in enumerate(matched_idx):
cid = valid_ids[mi]
tr = mem.traces[cid]
w = mem.current_weight(tr, now=now)
if w >= self.config.min_weight:
matches.append(tr)
weights.append(w)
sims.append(float(all_sims[j]))
if not matches:
return Retrieval(query=query, config=self.config,
matches=[], sims=[], weights=[], result=None)
# optional top-k by similarity
if self.config.topk is not None and len(matches) > self.config.topk:
order = np.argsort(sims)[::-1][: self.config.topk]
matches = [matches[i] for i in order]
sims = [sims[i] for i in order]
weights = [weights[i] for i in order]
return Retrieval(
query=query,
config=self.config,
matches=matches,
sims=sims,
weights=weights,
result=None,
)
def __call__(self, mem: Memory, query: HV) -> HV | None:
"""Compute ``Phi(query)`` — the reconstructed value (or None)."""
ret = self.retrieve(mem, query)
if not ret.matches:
return None
values = [tr.value for tr in ret.matches]
if self.config.sharpness <= 0.0:
# plain weighted bundle: majority of weighted signs
res = bundle(values, weights=ret.weights, deterministic=True)
else:
# similarity-temperatured weighting
temps = np.exp(self.config.sharpness * np.asarray(ret.sims))
eff = (np.asarray(ret.weights) * temps).tolist()
res = bundle(values, weights=eff, deterministic=True)
ret.result = res
return res
# ----------------------------------------------- parameter reconstruction
def reconstruct_param(self, mem: Memory, x: HV, state: HV) -> HV | None:
"""Axiome 4: ``W_t = Phi(bind(x, s_t))``.
The "weight" for input ``x`` in internal state ``s`` is reconstructed
on the fly from memory — it does not exist before this call and is not
stored after.
"""
from .hv import bind as _bind
return self(mem, _bind(x, state))
|