| """Predictive state: a model-based agent over hyperdimensional states. |
| |
| This is the layer the shading work was building toward — not pixels, but |
| *states*, and the two questions an agent actually has: |
| |
| 1. **What follows?** A transition is a bound pair, stored holographically per |
| action: ``table[action]`` is a :class:`~hdc.memory_modes.HolographicMemory` |
| keyed by state whose value is the next state. Prediction is unbind plus |
| cleanup over the state vocabulary — the same two primitives as everything |
| else here, with a similarity score as the confidence. |
| 2. **What should I do?** Planning is beam search over that model, scored by |
| ``sim(predicted, goal)``. No gradients, no sampling: a breadth-limited |
| lookahead whose every node is a vector compare. |
| |
| The measured quantity that makes this an *agent* rather than a lookup table is |
| **surprise**: ``1 − sim(predicted, actual)``. A trace of surprise over an |
| episode is the temporally measured loss. Rising surprise means the model is |
| out of distribution; that signal is what a self-correcting loop should act on — |
| re-plan, or learn the transition that was wrong. |
| |
| Honest limits: state identity is exact symbol cleanup, so the vocabulary is |
| finite; the model is first-order (one action of lookahead per step, not a |
| temporal stack); and beam search is exponential in horizon before the beam |
| width caps it. It is a small, inspectable world model, not an LLM. It is, |
| however, the part an LLM wrapper would call: *predict, score, choose*. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Callable, Iterable, Sequence |
|
|
| from .hypervec import DEFAULT_DIM, hvc, sim |
| from .memory_modes import HolographicMemory |
|
|
|
|
| class PredictiveModel: |
| """First-order transition model: ``(state, action) -> next state``.""" |
|
|
| def __init__(self, d: int = DEFAULT_DIM, alphabet: Sequence[int] | None = None): |
| self.d = d |
| self.alphabet = list(alphabet or []) |
| self.tables: dict[str, HolographicMemory] = {} |
| self.transitions = 0 |
|
|
| def _table(self, action: str) -> HolographicMemory: |
| table = self.tables.get(action) |
| if table is None: |
| table = HolographicMemory(d=self.d, alphabet=self.alphabet) |
| self.tables[action] = table |
| return table |
|
|
| def add_symbol(self, state: int) -> None: |
| """Add a state to the cleanup codebook, so prediction can resolve it. |
| |
| Without a codebook, unbinding returns the raw probe (value plus |
| crosstalk) and nothing ever compares equal to a stored state. Callers |
| that discover states as they go — a ring controller — add them here. |
| """ |
| if state not in self.alphabet: |
| self.alphabet.append(state) |
| for table in self.tables.values(): |
| if state not in table.alphabet: |
| table.alphabet.append(state) |
|
|
| def learn(self, state: int, action: str, next_state: int) -> None: |
| """Record one observed transition; repeated pairs add evidence.""" |
| self.add_symbol(state) |
| self.add_symbol(next_state) |
| table = self._table(action) |
| table.record(self._key(state), next_state) |
| self.transitions += 1 |
|
|
| @staticmethod |
| def _key(state: int) -> str: |
| """Exact key for a state. |
| |
| Exact keys keep unrelated contexts from interfering in the trace. The |
| cost is that a *similar* state is a miss, so :meth:`predict` follows up |
| with a snap to the nearest known key (see ``snap``) — precision on the |
| exact path, graded recall on the fallback. |
| """ |
| return "state/%x" % (state & 0xFFFFFFFF) |
|
|
| def predict(self, state: int, action: str, |
| snap: float = 0.5) -> tuple[int | None, float]: |
| """Predicted next state and cleanup confidence; ``(None, 0.0)`` if unseen. |
| |
| If the exact lookup is weak, the query is **snapped** to the nearest |
| trained key before unbinding. That is what lets a context that was |
| never seen still retrieve a transition: a paraphrase shares structure |
| with the trained keys, and key cleanup is the operation that exploits |
| it. ``snap=0`` disables the fallback. |
| """ |
| table = self.tables.get(action) |
| if table is None or not table.keys: |
| return None, 0.0 |
| predicted, confidence = table.recall(self._key(state)) |
| if confidence >= snap: |
| return predicted, confidence |
| key_vector, key_score = table.nearest_key(state) |
| if key_vector is not None and key_score > snap: |
| value, snapped = table.recall_bound(key_vector) |
| if snapped > confidence: |
| return value, snapped |
| return predicted, confidence |
|
|
| def name(self, state: int | None) -> str: |
| """Readable symbol for a state: its index in the vocabulary, or a hex tag.""" |
| if state is None: |
| return "?" |
| for i, candidate in enumerate(self.alphabet): |
| if candidate == state: |
| return "s%d" % i |
| return "#%08x" % (state & 0xFFFFFFFF) |
|
|
| def surprise(self, state: int, action: str, actual: int) -> float: |
| """``1 − sim(predicted, actual)`` in ``[0, 2]``; 0 means the model nailed it.""" |
| predicted, _confidence = self.predict(state, action) |
| if predicted is None: |
| return 1.0 |
| return 1.0 - sim(predicted, actual, self.d) |
|
|
| def actions(self) -> list[str]: |
| return sorted(self.tables) |
|
|
| def plan( |
| self, |
| start: int, |
| goal: int, |
| actions: Sequence[str] | None = None, |
| horizon: int = 4, |
| beam: int = 4, |
| ) -> dict: |
| """Beam search for the action sequence whose projection is closest to ``goal``. |
| |
| Returns ``{"actions": [...], "projected": state, "score": float, |
| "depth": int, "expanded": int}``. An empty action list with score |
| ``1.0`` means the goal was already reached. |
| """ |
| actions = list(actions or self.actions()) |
| best_score = sim(start, goal, self.d) |
| best_path: list[str] = [] |
| best_state = start |
| frontier = [(best_score, start, [])] |
| expanded = 0 |
| for depth in range(1, horizon + 1): |
| candidates = [] |
| for _score, state, path in frontier: |
| for action in actions: |
| predicted, confidence = self.predict(state, action) |
| if predicted is None: |
| continue |
| expanded += 1 |
| score = 0.6 * sim(predicted, goal, self.d) + 0.4 * confidence |
| candidates.append((score, predicted, path + [action])) |
| if score > best_score: |
| best_score, best_path, best_state = score, path + [action], predicted |
| if not candidates: |
| break |
| candidates.sort(key=lambda item: item[0], reverse=True) |
| frontier = candidates[:beam] |
| return {"actions": best_path, "projected": best_state, "score": best_score, |
| "depth": len(best_path), "expanded": expanded} |
|
|
|
|
| class StateAgent: |
| """Perceive, predict, choose, act — with surprise as the feedback signal.""" |
|
|
| def __init__(self, model: PredictiveModel, actions: Sequence[str], goal: int, beam: int = 4): |
| self.model = model |
| self.actions = list(actions) |
| self.goal = goal |
| self.beam = beam |
| self.episode: list[dict] = [] |
|
|
| def run( |
| self, |
| start: int, |
| world: Callable[[int, str], int], |
| steps: int = 6, |
| horizon: int = 3, |
| ) -> list[dict]: |
| """Act in ``world`` for ``steps``; ``world(state, action)`` returns the next state. |
| |
| At each step the model plans, takes the first action of the plan, then |
| measures surprise against what actually happened. The plan is |
| recomputed every step, so surprise can redirect it immediately. |
| """ |
| state = start |
| self.episode = [] |
| for _ in range(steps): |
| if sim(state, self.goal, self.model.d) > 0.99: |
| break |
| plan = self.model.plan(state, self.goal, self.actions, horizon=horizon, beam=self.beam) |
| action = plan["actions"][0] if plan["actions"] else self.actions[0] |
| predicted, confidence = self.model.predict(state, action) |
| actual = world(state, action) |
| surprise = self.model.surprise(state, action, actual) |
| self.episode.append({ |
| "state": state, "action": action, "predicted": predicted, |
| "actual": actual, "surprise": surprise, "plan": plan["actions"], |
| "confidence": confidence, |
| }) |
| self.model.learn(state, action, actual) |
| state = actual |
| return self.episode |
|
|
| def loss(self) -> float: |
| """Mean surprise over the episode — the temporally measured loss.""" |
| if not self.episode: |
| return 0.0 |
| return sum(step["surprise"] for step in self.episode) / len(self.episode) |
|
|
| def report(self) -> str: |
| lines = ["step action predicted actual surprise plan"] |
| for i, step in enumerate(self.episode): |
| lines.append("%4d %-7s %-9s %-6s %8.3f %s" % ( |
| i, step["action"], self.model.name(step["predicted"]), |
| self.model.name(step["actual"]), step["surprise"], |
| "->".join(step["plan"]) or "-")) |
| lines.append("mean surprise %.3f over %d steps" % (self.loss(), len(self.episode))) |
| return "\n".join(lines) |
|
|
|
|
| |
| |
|
|
| def ring_world(size: int = 8, d: int = DEFAULT_DIM) -> tuple[list[int], dict[str, Callable[[int], int]]]: |
| """States on a ring with ``next``/``prev`` actions, plus the state vocabulary.""" |
| states = [hvc("world/state/%d" % i, d) for i in range(size)] |
| index = {s: i for i, s in enumerate(states)} |
| return states, { |
| "next": lambda s: states[(index[s] + 1) % size], |
| "prev": lambda s: states[(index[s] - 1) % size], |
| } |
|
|
|
|
| def trained_model(size: int = 8, d: int = DEFAULT_DIM, epochs: int = 1) -> tuple[PredictiveModel, list[int], dict]: |
| """Learn the ring world completely, so planning has a correct model.""" |
| states, world = ring_world(size, d) |
| model = PredictiveModel(d=d, alphabet=states) |
| for _ in range(epochs): |
| for state in states: |
| for action, fn in world.items(): |
| model.learn(state, action, fn(state)) |
| return model, states, world |
|
|
|
|
| def report(d: int = DEFAULT_DIM, size: int = 8) -> str: |
| """Printable predictive-state tour.""" |
| model, states, world = trained_model(size, d) |
| target = states[-2] |
| plan = model.plan(states[0], target, ["next", "prev"], horizon=4, beam=4) |
| lines = ["predictive state: %d states, %d transitions, %d actions" |
| % (len(states), model.transitions, len(model.actions()))] |
| lines.append("plan s0 -> s%d: %s (score %.2f, %d nodes)" |
| % (states.index(target), "->".join(plan["actions"]), plan["score"], plan["expanded"])) |
| agent = StateAgent(model, ["next", "prev"], goal=target, beam=4) |
| agent.run(states[0], lambda s, a: world[a](s), steps=4, horizon=3) |
| lines.append(agent.report()) |
| return "\n".join(lines) |
|
|
|
|
| if __name__ == "__main__": |
| print(report()) |
|
|
