File size: 6,963 Bytes
1f71c7d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
"""PALIMPSESTE — Character-level tokenizer with vocab save/load.

For the hypervectorial substrate, a character-level tokenizer is the natural
choice: each character gets a stable atomic HV (via the :class:`Encoder`),
and text is encoded as a *sequence* (order-sensitive bundle of role-bound
char HVs). This keeps the vocabulary tiny and the encoding lossless, while
letting the associative memory learn character n-gram transitions.

This is deliberately **not** a BPE/SentencePiece tokenizer — those introduce
an opaque compression layer that fights the substrate's symbolic transparency
goal. Char-level keeps everything inspectable: you can always decode a
hypervector back to its nearest known characters.

The tokenizer implements the minimal HF-like interface needed for
``save_pretrained`` / ``from_pretrained`` compatibility:
``encode``, ``decode``, ``save_vocabulary``, ``vocab_size``, ``__len__``.
"""

from __future__ import annotations

from dataclasses import dataclass, field
import json
from pathlib import Path
import numpy as np

from .hv import HV, random_hv
from .learner import Encoder

__all__ = ["CharTokenizer", "VOCAB_SPECIAL"]


# Special tokens
VOCAB_SPECIAL = ["<pad>", "<bos>", "<eos>", "<unk>"]
PAD, BOS, EOS, UNK = 0, 1, 2, 3


@dataclass
class CharTokenizer:
    """Character-level tokenizer with a fixed atomic HV per character.

    The HVs are drawn from the shared :class:`Encoder` so that the same
    character always maps to the same hypervector, and the LM can bind them
    with positional roles to represent sequences.
    """

    encoder: Encoder
    # id -> char string (including special tokens at 0..3)
    id2char: list[str] = field(default_factory=lambda: list(VOCAB_SPECIAL))
    # char string -> id
    char2id: dict[str, int] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.char2id:
            self.char2id = {c: i for i, c in enumerate(self.id2char)}
        # pre-register atom HVs for all known chars (lazy otherwise)
        for ch in self.id2char:
            self._char_hv(ch)

    # ----------------------------------------------------------- properties
    @property
    def vocab_size(self) -> int:
        return len(self.id2char)

    def __len__(self) -> int:
        return self.vocab_size

    @property
    def pad_token_id(self) -> int:
        return PAD

    @property
    def bos_token_id(self) -> int:
        return BOS

    @property
    def eos_token_id(self) -> int:
        return EOS

    # ----------------------------------------------------------- vocab build
    def build_vocab(self, text: str) -> None:
        """Extend the vocabulary with every character in ``text``."""
        for ch in text:
            if ch not in self.char2id:
                self.char2id[ch] = len(self.id2char)
                self.id2char.append(ch)
                self._char_hv(ch)  # materialize the atom

    # ----------------------------------------------------------- HV lookup
    def _char_hv(self, ch: str) -> HV:
        """Get the atomic HV for a character (creates it lazily)."""
        return self.encoder.encode_str(f"__char__{ch}")

    def token_hv(self, token_id: int) -> HV:
        """Get the HV for a token id."""
        return self._char_hv(self.id2char[token_id])

    # ----------------------------------------------------------- encode/decode
    def encode(self, text: str, add_bos: bool = False, add_eos: bool = False) -> list[int]:
        ids = []
        if add_bos:
            ids.append(BOS)
        for ch in text:
            ids.append(self.char2id.get(ch, UNK))
        if add_eos:
            ids.append(EOS)
        return ids

    def decode(self, ids: list[int]) -> str:
        out = []
        for i in ids:
            if i in (PAD, BOS, EOS):
                continue
            if 0 <= i < len(self.id2char):
                out.append(self.id2char[i])
            # else skip unknown
        return "".join(out)

    # ----------------------------------------------------------- sequence HV
    def encode_text_to_hv(self, text: str) -> HV:
        """Encode a string into a single order-sensitive hypervector.

        Each character is bound with its positional role, then bundled.
        """
        ids = self.encode(text)
        hvs = [self.token_hv(i) for i in ids]
        if not hvs:
            return self.token_hv(PAD)
        return self.encoder.encode_sequence(hvs)

    def encode_context(self, ids: list[int], window: int | None = None,
                       start_pos: int = 0) -> HV:
        """Encode a token-id context (window of recent tokens) into one HV.

        This is the ``s_t`` state used by the LM: the last ``window`` tokens,
        each bound with its positional role, bundled.

        Roles are **absolute positions modulo window** (not relative indices).
        This means the token at absolute position ``p`` gets role ``p % window``,
        so the state HV is identical whether computed incrementally or from
        scratch — essential for associative retrieval consistency between
        training and inference.

        Parameters
        ----------
        ids : list of token ids
        window : context window size (default: all)
        start_pos : absolute position of the *first* id in ``ids`` (default 0).
            Used when encoding a suffix of a longer sequence so the role
            assignment matches the absolute positions.
        """
        if window is not None:
            # keep the last `window` ids, but track their absolute positions
            if len(ids) > window:
                ids = ids[-window:]
                start_pos = start_pos + (len(ids) - window) if start_pos else 0
        hvs = [self.token_hv(i) for i in ids]
        if not hvs:
            return self.token_hv(PAD)
        # bind each token with role(start_pos + i) % window
        from .hv import bind, bundle, random_hv
        bound = []
        for i, x in enumerate(hvs):
            role_idx = (start_pos + i) % window if window else i
            role_hv = self.encoder._role(role_idx)
            bound.append(bind(x, role_hv))
        return bundle(bound, rng=self.encoder.rng, deterministic=True)

    # ----------------------------------------------------------- save/load
    def save_vocabulary(self, path: str | Path) -> tuple[str]:
        """Save the vocab as ``vocab.json`` (HF-compatible name)."""
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        with open(p, "w", encoding="utf-8") as f:
            json.dump({"id2char": self.id2char}, f, indent=2, ensure_ascii=False)
        return (str(p),)

    @classmethod
    def load_vocabulary(cls, path: str | Path, encoder: Encoder) -> "CharTokenizer":
        p = Path(path)
        with open(p, "r", encoding="utf-8") as f:
            d = json.load(f)
        id2char = d["id2char"]
        tok = cls(encoder=encoder, id2char=id2char)
        return tok