hdcstorage / hdc /assistant.py
AEUPH's picture
HDCSTORAGE: hyperdimensional computer, webOS and agentic MCP layer (stdlib-only Python)
ad474c9 verified
Raw History Blame Contribute Delete
9.55 kB
"""The assistant: TextRing encoding + predictive model + real tool calls.
Honest about what it is. This is **not** an LLM. It has three parts, each of
which already existed in the project:
1. **Encoding.** An utterance is a bundle of its word vectors —
:mod:`hdc.text` n-grams by default, or the semantic encoder from
:mod:`hdc.semantic` when the model file is present. That bundle is the
utterance's hypervector.
2. **Classification.** A :class:`~hdc.state.PredictiveModel` maps utterance
vectors to intent vectors, with key snapping so a paraphrase can resolve to
a trained intent. The output is an intent plus a cleanup similarity, and
the similarity is reported to the user rather than hidden.
3. **Agency.** An intent names a kernel command. The assistant *runs it* and
returns what the machine actually printed, so a reply can carry a decoded
instruction listing, a run result, or a layer report instead of prose.
That structure is why it is useful inside this machine and also why it is
limited: the intent set is fixed at construction, the encoder knows nothing a
half-second of training cannot teach, and when no intent clears the threshold
it says so instead of inventing an answer.
"""
from __future__ import annotations
from typing import Sequence
from .hypervec import DEFAULT_DIM, bundle, hvc, sim
from .state import PredictiveModel
from .text import encode_word, tokenize
#: intent -> (example utterances, kernel command to run)
INTENTS: dict[str, tuple[list[str], str]] = {
"greet": (["hello", "hi there", "good morning", "hey", "greetings"],
""),
"help": (["what can you do", "show me the commands", "help me", "list commands"],
"help"),
"assemble": (["assemble a program", "compile this assembly", "build the demo program",
"assemble code"], "asm"),
"run": (["run the program", "execute it", "start the program", "run the code now"],
"run"),
"step": (["step one instruction", "single step", "advance the machine",
"step the cpu"], "step 1"),
"registers": (["show the registers", "what is in the registers", "register file",
"print regs", "what registers exist", "register contents",
"dump registers"], "regs"),
"disassemble": (["disassemble the program", "show the next instructions",
"what is the next instruction"], "disasm"),
"storage": (["show memory", "what is in storage", "dump the cells", "list cells"],
"mem"),
"status": (["what is the status", "machine status", "how many cells",
"show statistics"], "stat"),
"isa": (["show the instruction set", "what opcodes exist", "list opcodes",
"isa reference"], "isa"),
"shader": (["run the shader demo", "show the image pipeline", "edges and depth",
"shader report"], "shader"),
"ring": (["run the ring controller", "past present future", "show the history ledger",
"ring report"], "ring"),
"state": (["show the predictive state", "plan a path", "state report",
"what does the model predict"], "state"),
"memory_modes": (["photographic memory", "holographic recall", "memory modes report",
"how does recall work"], "memory"),
"topology": (["show the threshold topology", "component analysis", "self healing report",
"topology report"], "topology"),
"text": (["encode some text", "text encoder report", "character ngrams",
"how do you encode words"], "text"),
"semantic": (["semantic vectors", "word meanings", "embedding report",
"how close are synonyms"], "semantic"),
"boot": (["boot the machine", "run the bios", "start over", "reboot"],
"boot"),
"tools": (["show the tools", "what tools are registered", "tool registry"],
"tools"),
"clear": (["clear the screen", "wipe the console", "cls"], "clear"),
"thanks": (["thanks", "thank you", "nice", "great work"], ""),
}
#: Function words carry no intent and, bundled, they dominate the vector —
#: every utterance shares them, so prototypes collide on "the" and "of".
STOPWORDS = {
"the", "a", "an", "me", "my", "is", "are", "am", "was", "what", "how",
"do", "does", "it", "this", "that", "these", "those", "to", "of", "in",
"on", "for", "and", "or", "now", "some", "please", "can", "could", "you",
"i", "we", "us", "with", "at", "be", "there", "here", "up",
}
REPLIES = {
"greet": "Kernel is up. I can assemble, run, step and inspect — say what you want.",
"thanks": "Understood.",
"": "",
}
class Assistant:
"""A small, inspectable assistant over the kernel's own commands."""
def __init__(self, kernel, d: int = DEFAULT_DIM, encoder=None, threshold: float = 0.42):
self.kernel = kernel
self.d = d
self.encoder = encoder
# 0.42 keeps genuine paraphrases (measured 0.50-0.84) and rejects
# unrelated input (measured <= 0.38); reported, not tuned per-query
self.threshold = threshold
self.intent_vectors: dict[str, int] = {}
self.history: list[dict] = []
self.trained = 0
# One *prototype* per intent: the majority bundle of its examples.
# A single transition table holding all 82 utterances was measured at
# 0.05-0.09 confidence — past the majority trace's capacity — while a
# centroid of 4-6 members stays at ~0.5. Same rule as PrototypeNet.
self.prototypes: dict[str, int] = {}
for name, (examples, _command) in INTENTS.items():
self.intent_vectors[name] = hvc("intent/" + name, d)
self.prototypes[name] = bundle([self.encode(text) for text in examples], d)
self.trained += len(examples)
# kept for conversational follow-ups ("do that again")
self.model = PredictiveModel(d=d, alphabet=list(self.intent_vectors.values()))
for name, (examples, _command) in INTENTS.items():
for text in examples:
self.model.learn(self.encode(text), "intent", self.intent_vectors[name])
# ----------------------------------------------------------- encoding
def word(self, token: str) -> int:
if self.encoder is not None:
return self.encoder.encode(token)
return encode_word(token, self.d)
def encode(self, text: str) -> int:
"""An utterance as one bundled hypervector, content words only."""
tokens = [t for t in tokenize(text) if t not in STOPWORDS]
if not tokens:
tokens = tokenize(text) or [text.strip().lower()]
return bundle([self.word(token) for token in tokens], self.d)
# ------------------------------------------------------- classification
def classify(self, text: str) -> tuple[str, float]:
"""Nearest intent prototype and its similarity.
The score is a plain ``sim`` against a centroid, so it means the same
thing everywhere else in this library: 0.5 is "most of the members are
there", 0.05 is "unrelated". No softmax is applied to make it look
like a probability it is not.
"""
vector = self.encode(text)
best, best_score = "", -2.0
for name, prototype in self.prototypes.items():
score = sim(vector, prototype, self.d)
if score > best_score:
best, best_score = name, score
return best, best_score
# --------------------------------------------------------------- chat
def chat(self, text: str, execute=None) -> dict:
"""Classify, act, and return both the reply and what was executed.
``execute`` overrides how the named command is run. The autonomy layer
passes a cache-backed runner, so an answer the machine already predicted
costs a dict lookup instead of a kernel round trip; the default is the
kernel, so nothing changes for every other caller.
"""
intent, confidence = self.classify(text)
record = {"input": text, "intent": intent, "confidence": round(confidence, 3)}
if not intent or confidence < self.threshold:
record["reply"] = ("No intent cleared the threshold (best %.2f, need %.2f). "
"Try 'help' or one of the pipeline commands."
% (confidence, self.threshold))
record["output"] = ""
record["executed"] = ""
self.history.append(record)
return record
command = INTENTS[intent][1]
output = ""
if command:
output = (execute or self.kernel.execute)(command)
reply = REPLIES.get(intent) or ("Running '%s'. Confidence %.2f." % (command, confidence))
record.update({"reply": reply, "output": output, "executed": command})
self.history.append(record)
return record
# -------------------------------------------------------------- report
def report(self) -> str:
lines = ["assistant: %d intents, %d utterances trained, encoder=%s"
% (len(INTENTS), self.trained,
"semantic" if self.encoder is not None else "n-gram")]
for text in ("run the program", "show me the registers", "tell me a joke"):
intent, confidence = self.classify(text)
lines.append(" %-28s -> %-12s %.2f" % (text, intent or "(none)", confidence))
return "\n".join(lines)