| """Idle, learning, readahead: the assistant's autonomous mode. |
| |
| The assistant on its own is reactive — it answers what it is asked. This module |
| adds the three behaviours that make it *predictive* instead, and each one is a |
| measured quantity rather than a mood: |
| |
| **Idle.** Between utterances there is time the machine would otherwise waste. |
| An idle tick spends it on work that is useful whichever way the conversation |
| goes: replaying history, computing what comes next, and pre-running the |
| read-only inspection commands that are most likely to be asked for. |
| |
| **Learning.** Every turn leaves an intent. The sequence of intents is a trace |
| of behaviour, and ``(previous, next)`` pairs are learned into a |
| :class:`~hdc.state.PredictiveModel` — the same holographic transition table the |
| state layer uses. Idle *replays* that trace, and replay is not decorative: a |
| majority trace's cleanup confidence rises with repeated evidence, so a cycle |
| that was noisy after one lap sharpens on the next. Measured: see |
| :meth:`AutonomousMode.report`. |
| |
| **Readahead.** From the last intent, predict the next ``horizon`` intents by |
| chaining the transition model, each hop cleaned up against the intent codebook |
| with its own similarity. ``PAST (trace) -> PRESENT (last intent) -> FUTURE |
| (readahead) -> PAST (graded on the next turn)``: every prediction is stored and |
| graded when the actual next utterance is classified. That grading is the |
| learning signal, and its rate is the number the report leads with. |
| |
| **Autonomy, bounded.** Only commands on :data:`SAFE_COMMANDS` may be executed |
| without being asked, with a per-idle-call budget and a confidence floor. They |
| are read-only inspections, so the worst case of a wrong prediction is wasted |
| cycles, never a changed machine. When the user does ask for one of them, the |
| answer is already computed — which is the point: readahead trades idle time for |
| response time. |
| |
| Honest limits: intents are a fixed set, so this predicts *which* command comes |
| next and not what the user will say; a first-order chain has no notion of a |
| conversation's subject, only of its recent shape; and the confidence numbers are |
| cleanup similarities, not probabilities. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import time |
| from typing import Sequence |
|
|
| from .hypervec import DEFAULT_DIM, bundle, hvc, permute |
| from .state import PredictiveModel |
|
|
| |
| |
| |
| SAFE_COMMANDS = frozenset({ |
| "help", "isa", "stat", "regs", "mem", "disasm", "tools", "shader", "ring", |
| "state", "memory", "topology", "text", "semantic", |
| }) |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| ACT_FLOOR = 0.25 |
|
|
|
|
| class AutonomousMode: |
| """Idle ticks, transition learning, and graded readahead over the assistant.""" |
|
|
| def __init__(self, assistant, *, d: int | None = None, horizon: int = 3, |
| order: int = 1, budget: int = 4, act_floor: float = ACT_FLOOR, |
| enabled: bool = False): |
| self.assistant = assistant |
| self.d = d or assistant.d |
| self.horizon = horizon |
| self.order = order |
| self.budget = budget |
| self.act_floor = act_floor |
| self.enabled = enabled |
|
|
| self.intents: dict[str, int] = dict(assistant.intent_vectors) |
| self._names: dict[int, str] = {vector: name for name, vector in self.intents.items()} |
| self.model = PredictiveModel(d=self.d, alphabet=list(self.intents.values())) |
| self.sequence: list[str] = [] |
| self._learned = 0 |
|
|
| self.readahead: list[dict] = [] |
| self.graded: int = 0 |
| self.hits: int = 0 |
| self.ticks: int = 0 |
| self.replays: int = 0 |
| self.actions: list[dict] = [] |
| self.cache: dict[str, dict] = {} |
| self.served: int = 0 |
| self.ms_saved: float = 0.0 |
| self.at_floor_skips: int = 0 |
| self.log: list[dict] = [] |
|
|
| |
| def _key(self, window: Sequence[int]) -> int: |
| """Context for an ``order``-long window of intent vectors. |
| |
| Position-permuted so ``help->run`` and ``run->help`` are different keys; |
| with ``order=1`` it collapses to the intent itself, which is what a |
| deterministic command cycle needs. |
| """ |
| if len(window) == 1: |
| return window[0] |
| return bundle([permute(vector, -i, self.d) for i, vector in enumerate(reversed(window))], |
| self.d) |
|
|
| def _window(self) -> list[int]: |
| return [self.intents[name] for name in self.sequence[-self.order:]] |
|
|
| def learn_pending(self, epochs: int = 1) -> int: |
| """Learn every pair not yet recorded, ``epochs`` times over. |
| |
| Replay is genuine training for a majority trace: the same pair recorded |
| again increases its weight against crosstalk, so cleanup confidence — and |
| therefore prediction accuracy — improves with repetition. |
| """ |
| start = max(self._learned, self.order - 1) |
| learned = 0 |
| for _ in range(max(1, epochs)): |
| for i in range(start, len(self.sequence) - 1): |
| window = [self.intents[name] for name in self.sequence[i - self.order + 1:i + 1]] |
| self.model.learn(self._key(window), "next", self.intents[self.sequence[i + 1]]) |
| learned += 1 |
| self.replays += 1 |
| self._learned = max(self._learned, len(self.sequence) - 1) |
| return learned |
|
|
| def observe(self, intent: str, *, learn: bool = True) -> dict: |
| """Record one observed intent and grade the prediction made for it. |
| |
| Returns the grading: which intent was predicted, whether it was right, |
| and the confidence that prediction carried before the answer was known. |
| """ |
| expected = self.readahead[0] if self.readahead else None |
| verdict = {"predicted": (expected or {}).get("intent"), "actual": intent, |
| "confidence": (expected or {}).get("confidence", 0.0), |
| "hit": bool(expected and expected.get("intent") == intent)} |
| if expected is not None: |
| self.graded += 1 |
| if verdict["hit"]: |
| self.hits += 1 |
| self.sequence.append(intent) |
| if learn: |
| self.learn_pending() |
| self.readahead = self.predict() |
| return verdict |
|
|
| |
| def name_of(self, vector: int | None) -> str | None: |
| return self._names.get(vector) if vector is not None else None |
|
|
| def predict(self, horizon: int | None = None) -> list[dict]: |
| """Chain the transition model forward from the last observed intent. |
| |
| Each hop is cleaned up against the intent codebook; ``path`` is the |
| running confidence product, ``confidence`` the raw hop similarity, so a |
| three-step guess cannot look as certain as a one-step one. |
| """ |
| horizon = self.horizon if horizon is None else horizon |
| window = self._window() |
| if not window: |
| return [] |
| state = self._key(window) |
| out: list[dict] = [] |
| seen: set[str] = set() |
| path = 1.0 |
| for depth in range(1, horizon + 1): |
| predicted, confidence = self.model.predict(state, "next") |
| name = self.name_of(predicted) |
| if name is None or confidence <= 0.0: |
| break |
| path *= confidence |
| out.append({"intent": name, "command": self._command(name), |
| "confidence": round(confidence, 3), "path": round(path, 3), |
| "depth": depth, "cached": self._command(name) in self.cache}) |
| seen.add(name) |
| |
| |
| state = predicted |
| if len(seen) == len(self.intents): |
| break |
| return out |
|
|
| def _command(self, intent: str) -> str: |
| from .assistant import INTENTS |
|
|
| return INTENTS.get(intent, ([], ""))[1] |
|
|
| |
| def tick(self) -> dict: |
| """One idle cycle: replay, predict, and pre-execute what is safe.""" |
| self.ticks += 1 |
| learned = self.learn_pending() |
| self.readahead = self.predict() |
| acted = [] |
| if self.enabled: |
| acted = self._act() |
| entry = {"tick": self.ticks, "learned": learned, "acted": acted, |
| "readahead": [item["intent"] for item in self.readahead[:3]]} |
| self.log.append(entry) |
| return entry |
|
|
| def idle(self, ticks: int = 1) -> list[dict]: |
| return [self.tick() for _ in range(max(0, ticks))] |
|
|
| def _act(self, budget: int | None = None) -> list[dict]: |
| """Pre-run the predicted read-only commands, within the budget.""" |
| budget = self.budget if budget is None else budget |
| acted: list[dict] = [] |
| for prediction in self.readahead: |
| if len(acted) >= budget: |
| break |
| command = prediction["command"] |
| if not command or command.split()[0] not in SAFE_COMMANDS: |
| continue |
| if command in self.cache: |
| continue |
| if prediction["confidence"] < self.act_floor: |
| self.at_floor_skips += 1 |
| continue |
| started = time.perf_counter() |
| output = self.assistant.kernel.execute(command) |
| elapsed = (time.perf_counter() - started) * 1000.0 |
| self.cache[command] = {"output": output, "ms": elapsed, |
| "confidence": prediction["confidence"]} |
| record = {"command": command, "intent": prediction["intent"], |
| "confidence": prediction["confidence"], "ms": round(elapsed, 2)} |
| acted.append(record) |
| self.actions.append(record) |
| return acted |
|
|
| |
| def chat(self, text: str) -> dict: |
| """``Assistant.chat`` with the readahead cache in front of execution.""" |
| intent, confidence = self.assistant.classify(text) |
| command = self._command(intent) if intent else "" |
| served = command and command in self.cache and confidence >= self.assistant.threshold |
| if served: |
| cached = self.cache[command] |
| self.served += 1 |
| self.ms_saved += cached["ms"] |
| record = self.assistant.chat(text, execute=self._executor if served else None) |
| if served: |
| record["output"] = self.cache[command]["output"] |
| record["cached"] = True |
| record["saved_ms"] = round(self.cache[command]["ms"], 2) |
| verdict = self.observe(intent) if intent else None |
| record["graded"] = verdict |
| if self.enabled: |
| self.tick() |
| return record |
|
|
| def _executor(self, command: str) -> str: |
| cached = self.cache.get(command) |
| if cached is not None: |
| return cached["output"] |
| return self.assistant.kernel.execute(command) |
|
|
| |
| def chance(self) -> float: |
| """Cleanup similarity a random symbol scores against the intent codebook.""" |
| return 1.0 / max(1.0, len(self.intents) ** 0.5) |
|
|
| def stats(self) -> dict: |
| accuracy = (self.hits / self.graded) if self.graded else 0.0 |
| return { |
| "enabled": self.enabled, "ticks": self.ticks, "replays": self.replays, |
| "observed": len(self.sequence), "transitions": self.model.transitions, |
| "graded": self.graded, "hits": self.hits, "accuracy": round(accuracy, 3), |
| "chance": round(self.chance(), 3), |
| "readahead": self.readahead, "cached_commands": sorted(self.cache), |
| "autonomous_actions": len(self.actions), "served_from_readahead": self.served, |
| "ms_saved": round(self.ms_saved, 1), "act_floor": self.act_floor, |
| "budget": self.budget, |
| } |
|
|
| def report(self) -> str: |
| stats = self.stats() |
| lines = ["autonomy: %s, horizon %d, order %d, %.2f confidence floor" |
| % ("enabled" if self.enabled else "disabled", |
| self.horizon, self.order, self.act_floor)] |
| lines.append(" idle ticks %-3d replays %-3d transitions %-3d observed %d" |
| % (stats["ticks"], stats["replays"], stats["transitions"], |
| stats["observed"])) |
| lines.append(" readahead graded %d, correct %d -> %.1f%% next-intent accuracy " |
| "(chance %.2f)" |
| % (stats["graded"], stats["hits"], 100.0 * stats["accuracy"], |
| stats["chance"])) |
| if self.readahead: |
| lines.append(" current lookahead: " + " -> ".join( |
| "%s(%.2f)" % (item["intent"], item["confidence"]) for item in self.readahead)) |
| else: |
| lines.append(" current lookahead: none (observation needed first)") |
| lines.append(" autonomous actions %d (%d skipped under the floor), cached %d, " |
| "served %d, %.1f ms saved" |
| % (stats["autonomous_actions"], self.at_floor_skips, |
| len(self.cache), stats["served_from_readahead"], stats["ms_saved"])) |
| return "\n".join(lines) |
|
|
|
|
| |
| SESSION = ( |
| "show the statistics", "show the registers", "show the instruction set", |
| "what can you do", |
| "show the statistics", "show the registers", "show the instruction set", |
| "what can you do", |
| "show the statistics", "show the registers", "show the instruction set", |
| "what can you do", |
| ) |
|
|
|
|
| def simulate(session: Sequence[str] = SESSION, *, d: int = 2000, encoder=None, |
| enabled: bool = True, warmup: int = 1) -> dict: |
| """Run a scripted session through idle + chat and report what it learned. |
| |
| The point of the simulation is that the numbers in :meth:`AutonomousMode.report` |
| are measured on a conversation, not asserted. ``warmup`` idle ticks run |
| before the first utterance, so the machine starts in a cold state and has to |
| earn its accuracy. |
| """ |
| from .assistant import Assistant |
| from .terminal import Kernel |
|
|
| kernel = Kernel(d=d) |
| assistant = Assistant(kernel, d=d, encoder=encoder) |
| mode = AutonomousMode(assistant, d=d, enabled=enabled) |
| mode.idle(warmup) |
|
|
| per_turn: list[dict] = [] |
| for text in session: |
| record = mode.chat(text) |
| per_turn.append({"input": text, "intent": record.get("intent"), |
| "graded": record.get("graded"), |
| "cached": record.get("cached", False), |
| "saved_ms": record.get("saved_ms", 0.0)}) |
| half = len(per_turn) // 2 |
| early = [t for t in per_turn[:half] if t["graded"]] |
| late = [t for t in per_turn[half:] if t["graded"]] |
| return { |
| "mode": mode, "turns": per_turn, |
| "early_accuracy": _rate(early), "late_accuracy": _rate(late), |
| "stats": mode.stats(), "report": mode.report(), |
| } |
|
|
|
|
| def _rate(turns: Sequence[dict]) -> float: |
| if not turns: |
| return 0.0 |
| return sum(1 for turn in turns if turn["graded"]["hit"]) / len(turns) |
|
|
|
|
| def report(d: int = DEFAULT_DIM) -> str: |
| """The printable form: a simulated session, its accuracy and its savings.""" |
| from .semantic import load |
|
|
| encoder = load(d=d) |
| result = simulate(d=d, encoder=encoder) |
| lines = ["autonomy report (simulated session, %d utterances, encoder=%s)" |
| % (len(SESSION), "semantic" if encoder else "n-gram")] |
| lines.append(result["report"]) |
| lines.append(" next-intent accuracy: first half %.0f%% -> second half %.0f%%" |
| % (100.0 * result["early_accuracy"], 100.0 * result["late_accuracy"])) |
| for turn in result["turns"][-4:]: |
| graded = turn["graded"] or {} |
| lines.append(" %-26s -> %-10s predicted %-10s %s%s" |
| % (turn["input"][:26], turn["intent"], |
| graded.get("predicted") or "-", |
| "hit" if graded.get("hit") else "miss", |
| " [readahead]" if turn["cached"] else "")) |
| return "\n".join(lines) |
|
|
|
|
| if __name__ == "__main__": |
| print(report()) |
|
|
