palimseste/soft_phi.py

"""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)