Improve chat runtime and probe REPL
This commit is contained in:
parent
f8113a38ba
commit
4c3004c73a
5 changed files with 71 additions and 5 deletions
|
|
@ -1,21 +1,33 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from language_packs import OOVPolicy, load_pack, load_pack_entries
|
||||
from persona.motor import PersonaMotor
|
||||
from field.state import FieldState
|
||||
from session.context import SessionContext
|
||||
|
||||
_TOKEN_RE = re.compile(r"\w+", re.UNICODE)
|
||||
|
||||
|
||||
class ChatRuntime:
|
||||
def __init__(self, pack_id: str = "en_minimal_v1") -> None:
|
||||
manifest, manifold = load_pack(pack_id)
|
||||
entries = load_pack_entries(pack_id)
|
||||
self._manifest = manifest
|
||||
self._context = SessionContext(manifold, persona=PersonaMotor.identity())
|
||||
self._index_by_surface = {w: i for i, w in enumerate(self._context.vocab._words)}
|
||||
self._surface_by_fold = {e.surface.casefold(): e.surface for e in entries}
|
||||
self._pos_by_surface = {
|
||||
e.surface: (e.pos or e.part_of_speech or "X") for e in load_pack_entries(pack_id)
|
||||
e.surface: (e.pos or e.part_of_speech or "X") for e in entries
|
||||
}
|
||||
|
||||
def _tokenize(self, text: str) -> list[str]:
|
||||
tokens: list[str] = []
|
||||
for match in _TOKEN_RE.finditer(text):
|
||||
raw = match.group(0)
|
||||
tokens.append(self._surface_by_fold.get(raw.casefold(), raw))
|
||||
return tokens
|
||||
|
||||
def _apply_oov_policy(self, tokens: list[str]) -> list[str]:
|
||||
kept: list[str] = []
|
||||
for token in tokens:
|
||||
|
|
@ -41,12 +53,12 @@ class ChatRuntime:
|
|||
return out
|
||||
|
||||
def respond(self, text: str, max_tokens: int = 32) -> str:
|
||||
tokens = [t.strip() for t in text.split() if t.strip()]
|
||||
tokens = self._tokenize(text)
|
||||
filtered = self._apply_oov_policy(tokens)
|
||||
if not filtered:
|
||||
return ""
|
||||
self._context.ingest(filtered)
|
||||
node_idx = self._index_by_surface.get(filtered[0], 0)
|
||||
node_idx = self._context.vocab.index_of(filtered[0])
|
||||
self._context.state = FieldState(
|
||||
F=self._context.state.F,
|
||||
node=node_idx,
|
||||
|
|
|
|||
|
|
@ -17,7 +17,10 @@ def field_walk(seed: str, steps: int = 4) -> list[str]:
|
|||
|
||||
def main() -> None:
|
||||
while True:
|
||||
text = input("> ").strip()
|
||||
try:
|
||||
text = input("> ").strip()
|
||||
except EOFError:
|
||||
break
|
||||
if text in {"quit", "exit"}:
|
||||
break
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from chat.runtime import ChatRuntime
|
||||
from language_packs import load_pack
|
||||
|
||||
|
|
@ -24,3 +26,21 @@ def test_chat_runtime_responds_and_varies():
|
|||
assert a.strip()
|
||||
assert b.strip()
|
||||
assert a != b
|
||||
|
||||
|
||||
def test_chat_runtime_canonicalizes_case_and_punctuation():
|
||||
plain = ChatRuntime("en_minimal_v1").respond("what is light", max_tokens=8)
|
||||
punctuated = ChatRuntime("en_minimal_v1").respond("WHAT is light?", max_tokens=8)
|
||||
assert punctuated == plain
|
||||
|
||||
|
||||
def test_chat_runtime_fail_closed_pack_rejects_oov():
|
||||
runtime = ChatRuntime("he_logos_micro_v1")
|
||||
with pytest.raises(KeyError):
|
||||
runtime.respond("light", max_tokens=4)
|
||||
|
||||
|
||||
def test_vocab_manifold_exposes_public_word_index():
|
||||
_, manifold = load_pack("en_minimal_v1")
|
||||
idx = manifold.index_of("light")
|
||||
assert manifold.get_word_at(idx) == "light"
|
||||
|
|
|
|||
23
tests/test_probe_repl.py
Normal file
23
tests/test_probe_repl.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import builtins
|
||||
|
||||
from probe import repl
|
||||
|
||||
|
||||
def test_repl_exits_cleanly_on_stdin_eof(monkeypatch, capsys):
|
||||
inputs = iter(["light"])
|
||||
|
||||
def fake_input(prompt: str) -> str:
|
||||
try:
|
||||
return next(inputs)
|
||||
except StopIteration:
|
||||
raise EOFError
|
||||
|
||||
monkeypatch.setattr(builtins, "input", fake_input)
|
||||
|
||||
repl.main()
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "[field walk:" in out
|
||||
assert "light" in out
|
||||
|
|
@ -19,6 +19,7 @@ Normalization doctrine for this module:
|
|||
Indexed access:
|
||||
get_versor_at(idx) — returns a copy of the stored versor by integer index.
|
||||
get_word_at(idx) — returns the word string by integer index.
|
||||
index_of(word) — returns the integer index for a stored word.
|
||||
These are the primitives generation uses; VocabManifold does not build
|
||||
operators. Algebra builds operators. Vocab stores points.
|
||||
|
||||
|
|
@ -84,6 +85,13 @@ class VocabManifold:
|
|||
"""Return the word string at integer index."""
|
||||
return self._words[idx]
|
||||
|
||||
def index_of(self, word: str) -> int:
|
||||
"""Return the integer index for a stored word. Raises KeyError if missing."""
|
||||
try:
|
||||
return self._words.index(word)
|
||||
except ValueError:
|
||||
raise KeyError(f"Word '{word}' not in vocabulary.")
|
||||
|
||||
def nearest(self, F: np.ndarray, exclude_idx: int = -1) -> tuple[str, int]:
|
||||
"""
|
||||
Find the word whose versor is closest to F by CGA inner product.
|
||||
|
|
|
|||
Loading…
Reference in a new issue