本文章没有找到对应的语言版本

1. 引言

Tokenizer 是大模型的“输入接口”：同一段文本被切成怎样的 token，直接影响序列长度、训练/推理成本，甚至影响模型对边界与模式的学习。

在实践中，你可能会遇到这些现象：

  • 同一个单词，句首与句中切分不同（"Hello" vs " Hello"）。
  • 数字被切得很碎（"12345" 变成 "12" "34" "5"）。
  • 中文在英文模型里 token 更多，但并不是“不能编码”，而是更长。
  • 训练得到的 merges（合并规则）看起来难以理解：它究竟如何在推理中生效？

本文把这些现象统一到一条可复现的流水线：

特殊 token → 正则预分词（分块） → UTF-8 字节 → BPE 合并 → token id

同时给出一份“最小但完整”的 Python 参考实现（训练 / 编码 / 解码 / 保存与加载），帮助你把概念落到可运行代码上。

2. BPE 的基本原理

Byte Pair Encoding（BPE）最早来自数据压缩：反复把序列里最常见的相邻符号对合成一个新符号，从而缩短序列。迁移到 NLP 后，符号不再局限于“字符/词”，而是介于两者之间的 subword。

BPE 的目标可以概括为：

  • 高频片段被合并成更大的 token（序列更短、推理更快）。
  • 低频片段保持更细粒度（组合即可覆盖，缓解 OOV）。
  • 词表大小可控（几十 K 到十几万都常见）。

2.1 merges：有序的合并规则

BPE 训练的直接产物不是“分词词典”，而是一组有序 merges。每条规则可以写成：

$$ (a, b) \rightarrow ab $$

其中 $a$、$b$ 是相邻 token，$ab$ 是拼接后的新 token。更重要的是：merges 的顺序就是推理时的优先级（rank）。顺序改变，编码结果就会改变。

2.2 一个直观的小例子

以（仅用于说明思路的）字符序列为例：

1
2
3
4
5

初始: l o w  <space>  l o w e r
最频繁对: (l, o)  → 合并为 lo
得到: lo w  <space>  lo w e r
再次统计最频繁对: (lo, w) → 合并为 low
得到: low  <space>  low e r

这里的关键不是“合并成什么”，而是“合并的顺序会被保存下来”，推理时要以同样的顺序（或等价的 rank 逻辑）去复现。

3. 字节级 BPE：为什么它几乎成为默认选择

经典教程常从“字符级 + 单词结束符 </w>”讲起，这更贴近早期子词分词论文；但 GPT 系列等现代大模型更常使用 byte-level BPE（字节级 BPE）：

  1. 把文本按 UTF-8 编码为字节序列（每个字节在 0~255）。
  2. 初始词表固定就是这 256 个字节 token。
  3. 在此基础上学习 merges，把常见的字节片段合成更大的 token。

这种设计的直接好处是：任何 Unicode 都可编码。因为任何字符都能表示为 UTF-8 字节序列，而 0~255 的字节 token 永远在词表里，因此理论上不需要 <unk>。

3.1 Byte Fallback 到底是什么？

不少实现会同时提到“字节级词表”和“byte fallback”，但它们不是同一个概念：

  • 字节级 BPE：基础词表显式包含 0~255 字节 token，因此文本永远可表示。
  • Byte Fallback（字节回退）：当某个字符/片段在词表里找不到时，允许回退到“按字节拆开”的表示（具体行为因实现而异）。

在“256 字节起步”的设定下，fallback 很多时候是“自然成立”的：最终实在合不动，就保持为单字节 token。它的现实意义是：编码永远不会失败，但某些文本会更长。

4. 预分词（Pre-tokenization）：为什么正则很关键

如果把整段文本直接丢给 BPE，合并可能跨越不合理的边界（例如把逗号与单词粘连、跨越换行等），带来两个后果：

  • 词表被“标点组合”“空格组合”污染，浪费容量。
  • 切分不稳定，模型更难学习规律。

因此主流做法是：先用正则把文本切成 chunk，再对 每个 chunk 独立做 BPE 合并。

一个典型的预分词效果（示意）：

1
2

输入:  Hello, world! I'm learning.
输出: ["Hello", ",", " world", "!", " I", "'m", " learning", "."]

注意 "Hello" 与 " world"：前导空格被保留下来。这解释了一个常见现象："Hello" 与 " Hello" 可能对应不同 token。这不是 bug，而是一种把“词边界/位置信息”编码进 token 的设计。

4.1 一个常用的 GPT 风格 split pattern

下面的 pattern（示例）会分离缩写、字母串、数字、标点与空白，并尽量把“前导空格 + 内容”作为一个 chunk：

1

SPLIT_PATTERN = r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}+|\p{N}{1,2}| ?[^\s\p{L}\p{N}]++[\r\n]*|\s*[\r\n]|\s+(?!\S)|\s+"""

这类 split pattern 常见于 GPT 风格 tokenizer（例如 tiktoken 的实现思路）。 只保留关键解释：

  • '( ?i: ... )：把 "'m" "'t" "'re" 等缩写单独切出来。
  • ...\p{L}+：字母串（\p{L} 支持 Unicode 字母，中文也包含在内）。
  • \p{N}{1,2}：把数字按 1~2 位分组（OpenAI 官方实现为 1~3 位）；可减少“数字组合”对词表的占用，但会拉长数字序列。

注意：实现时请使用第三方 regex 库（Python 内置 re 不支持 \p{L}、++、?+ 等特性）。

5. 训练阶段：如何得到 merges

训练阶段可以概括为“在所有 chunk 内反复寻找最常见的相邻 token 对”。核心步骤如下：

  1. 初始化词表：256 个字节 token + 若干特殊 token。
  2. 对语料做正则预分词，得到很多 chunk。
  3. 每个 chunk 转为字节 token 序列（每字节一个 token）。
  4. 统计所有 chunk 中相邻 token 对的频率。
  5. 选出最频繁的 pair (a, b)，合并为新 token ab 并加入词表。
  6. 在所有 chunk 内应用同样的合并替换，继续迭代，直到达到目标词表大小或再也合不动。

两个容易踩坑的点：

  • 合并必须在 chunk 内进行：不要跨 chunk 边界合并，否则会破坏词边界/标点边界/换行边界。
  • merges 的顺序必须被保留：把 merges 想成“合并规则的优先级表”。训练时越早学到的合并越常见，排得越靠前（rank 越小）；推理时也必须先按这些靠前规则去合并，否则切分就会和训练时的优先级不一致。

6. 推理阶段：merges 如何生效（rank 贪心合并）

推理阶段的目标是：给定一个 chunk 的初始字节 token 序列，使用 merges 把它合并成最终 token 序列。

常见实现有两种：

  • 顺序扫描（直观但慢）：按 merges 的顺序逐条扫描并替换。
  • rank 贪心（工程常用）：把 merges 转成 pair -> rank，在当前序列中反复选择“rank 最小的可合并 pair”进行合并，直到不存在可合并 pair。

rank 贪心更贴近 BPE 语义：训练时每轮只合并一个 pair（当前最频繁）；推理时每轮合并一个 pair（当前 rank 最小，即 merges 列表最前的），直至无法合并。

7. 一份可运行的 Python 参考实现（训练 / 编码 / 解码 / 保存）

下面给出一份“最小但完整”的参考实现，便于复现本文的关键机制：

  • 字节级基础词表（256 字节）。
  • 正则预分词（chunk）。
  • chunk 内 BPE 合并（rank 贪心）。
  • 特殊 token 精确匹配（不参与合并）。
  • JSON 保存与加载（可复现、可部署）。
  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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304

# pip install regex loguru

import json
from collections import defaultdict
from dataclasses import dataclass, field

import regex
from loguru import logger

# tiktoken
SPLIT_PATTERN = r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}+|\p{N}{1,3}| ?[^\s\p{L}\p{N}]++[\r\n]*|\s*[\r\n]|\s+(?!\S)|\s+"""
PATTERN = regex.compile(SPLIT_PATTERN)


@dataclass
class TokenizerContext:
    vocab_size: int
    special_tokens: list[str] = field(default_factory=list)

    # post init
    merges: list[tuple[bytes, bytes]] = field(default_factory=list)
    id2token: dict[int, bytes] = field(init=False)
    token2id: dict[bytes, int] = field(init=False)

    def __post_init__(self):
        assert self.vocab_size >= 256
        self.token2id = {bytes([i]): i for i in range(256)}
        self.id2token = {i: bytes([i]) for i in range(256)}
        self._update_special_token()
        assert (
            len(self.token2id) <= self.vocab_size
        ), f"special_token 太多({len(self.id2token)})，导致词表已超过最大词表数量({self.vocab_size})"

    def update_token_map(self, pair: tuple[bytes, bytes]):
        new_token = pair[0] + pair[1]
        # 检查 new_token 是否已存在；如果已存在，会把token2id[new_token] 覆盖成新 id，
        # 同时 id2token 会留下旧 id 的“脏数据”，导致同一个 token 对应多个 id、以及后续 decode/encode 语义混乱。
        existing_id = self.token2id.get(new_token)
        if existing_id is not None:
            if self.id2token.get(existing_id) is None:
                self.id2token[existing_id] = new_token
            self.merges.append(pair)
            return

        new_id = len(self.token2id)
        self.token2id[new_token] = new_id
        self.id2token[new_id] = new_token
        self.merges.append(pair)

    def restore_token_map(self, merges: list[tuple[bytes, bytes]]):
        for pair in merges:
            self.update_token_map(pair)

    def _update_special_token(self):
        for token in self.special_tokens:
            token = token.encode("utf-8")
            self.token2id[token] = len(self.token2id)
            self.id2token[self.token2id[token]] = token

    @property
    def num_merges(self):
        return self.vocab_size - len(self.token2id)


class BPETokenizer:

    def __init__(self, **kwargs):
        self.ctx = TokenizerContext(**kwargs)

    def _iter_text_special_chunks(self, text: str):
        if not self.ctx.special_tokens:
            yield text, False
            return

        special_tokens = sorted(self.ctx.special_tokens, key=len, reverse=True)
        special_pattern = regex.compile(
            "|".join(regex.escape(tok) for tok in special_tokens)
        )

        last = 0
        for match in special_pattern.finditer(text):
            if match.start() > last:
                yield text[last : match.start()], False
            yield match.group(0), True
            last = match.end()
        if last < len(text):
            yield text[last:], False

    def train(self, corpus: list[str]):
        all_chunk_tokens = []
        for text in corpus:
            # 根据special_token进行分块
            for segment, is_special in self._iter_text_special_chunks(text):
                if is_special:
                    all_chunk_tokens.append([segment.encode("utf-8")])
                    continue

                # 对正常内容进行分块
                chunks = PATTERN.findall(segment)
                for chunk in chunks:
                    chunk_bytes = chunk.encode("utf-8")
                    tokens = [bytes([b]) for b in chunk_bytes]
                    all_chunk_tokens.append(tokens)

        for i in range(self.ctx.num_merges):
            # 统计所有 chunk 中的 token 对频率
            stats = defaultdict(int)
            for chunk in all_chunk_tokens:
                # 统计每个chunk中的频率
                chunk_stats = self.get_stats(chunk)
                for pair, count in chunk_stats.items():
                    stats[pair] += count

            if not stats:
                logger.warning(f"没有更多可合并的 token 对，在第 {i} 次迭代停止")
                break

            # 找到最频繁的 token 对
            best_pair = max(stats, key=stats.get)

            # 在每个 chunk 中独立合并这个 token 对
            all_chunk_tokens = [
                self.merge_tokens(chunk, best_pair) for chunk in all_chunk_tokens
            ]

            # 更新词汇表
            self.ctx.update_token_map(best_pair)

            # 日志输出
            if (i + 1) % 50 == 0 or i < 10:
                freq = stats[best_pair]
                new_token = best_pair[0] + best_pair[1]
                new_id = self.ctx.token2id[new_token]

                try:
                    token_str = new_token.decode("utf-8")
                except:  # noqa
                    token_str = new_token.hex()
                logger.info(
                    f"合并 #{i+1}: {best_pair[0]!r} + {best_pair[1]!r} "
                    f"-> {new_token!r} ({token_str}) (频率: {freq}, token_id: {new_id})"
                )

        logger.info(f"训练完成！最终词汇表大小: {len(self.ctx.token2id)}")
        logger.info(f"合并规则数量: {len(self.ctx.merges)}")
        logger.info(f"合并规则（前10）: {self.ctx.merges[:10]}")

    @staticmethod
    def get_stats(tokens: list[bytes]) -> dict[tuple[bytes, bytes], int]:
        """统计相邻 token 对的频率

        Args:
            tokens: token 列表

        Returns:
            {(token1, token2): 频率} 的字典
        """
        counts = defaultdict(int)
        for pair in zip(tokens[:-1], tokens[1:]):
            counts[pair] += 1
        return counts

    @staticmethod
    def merge_tokens(tokens: list[bytes], pair: tuple[bytes, bytes]) -> list[bytes]:
        """在 token 列表中合并指定的 token 对

        Args:
            tokens: 原始 token 列表
            pair: 要合并的 token 对 (token1, token2)

        Returns:
            合并后的 token 列表
        """
        new_tokens = []
        i = 0
        while i < len(tokens):
            # 检查是否匹配要合并的 pair
            if (
                i < len(tokens) - 1
                and tokens[i] == pair[0]
                and tokens[i + 1] == pair[1]
            ):
                new_tokens.append(pair[0] + pair[1])
                i += 2
            else:
                new_tokens.append(tokens[i])
                i += 1
        return new_tokens

    def encode(self, text: str) -> list[int]:
        """将文本编码为 token ID 序列

        编码过程：
        1. 使用正则表达式进行预分词
        2. 对每个片段应用 BPE 编码
        3. 返回所有 token ID

        Args:
            text: 输入文本

        Returns:
            token ID 列表
        """
        all_token_ids = []
        for segment, is_special in self._iter_text_special_chunks(text):
            if is_special:
                token = segment.encode("utf-8")
                token_id = self.ctx.token2id[token]
                all_token_ids.append(token_id)
                continue

            chunks = PATTERN.findall(segment)
            for chunk in chunks:
                chunk_bytes = chunk.encode("utf-8")
                token_ids = self.encode_chunk(chunk_bytes)
                all_token_ids.extend(token_ids)
        return all_token_ids

    def encode_chunk(self, chunk_bytes: bytes) -> list[int]:
        """对单个文本片段（已转换为字节）应用 BPE 编码

        Args:
            chunk_bytes: 文本片段的字节表示

        Returns:
            token ID 列表
        """
        # 初始化：每个字节作为一个 token
        tokens = [bytes([b]) for b in chunk_bytes]
        # 按顺序应用所有合并规则
        for pair in self.ctx.merges:
            tokens = self.merge_tokens(tokens, pair)

        # 转换为 ID
        token_ids = [self.ctx.token2id[token] for token in tokens]
        return token_ids

    def decode(self, token_ids: list[int]) -> str:
        """将 token ID 序列解码为文本

        Args:
            token_ids: token ID 列表

        Returns:
            解码后的文本
        """
        # 将 ID 转换为字节
        logger.info(f"token_ids: {token_ids}")
        token_bytes = [self.ctx.id2token[id] for id in token_ids]

        # 拼接所有字节
        all_bytes = b"".join(token_bytes)

        # 解码为字符串
        text = all_bytes.decode("utf-8", errors="replace")
        return text

    def save_vocab(self, filepath: str) -> None:
        """保存词汇表和合并规则到文件"""
        data = {
            "vocab_size": self.ctx.vocab_size,
            "merges": [(p[0].hex(), p[1].hex()) for p in self.ctx.merges],
            "special_tokens": self.ctx.special_tokens,
        }

        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

        logger.info(f"词汇表已保存到: {filepath}")

    def load_vocab(self, filepath: str) -> None:
        """从文件加载词汇表和合并规则"""
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)

        self.ctx = TokenizerContext(
            **dict(
                vocab_size=data["vocab_size"],
                special_tokens=data.get("special_tokens", []),
            )
        )
        # 重建词汇表
        merges = [(bytes.fromhex(p[0]), bytes.fromhex(p[1])) for p in data["merges"]]
        self.ctx.restore_token_map(merges)

        logger.info(f"词汇表已从 {filepath} 加载")


if __name__ == "__main__":
    corpus = [
        "Hello, world! I'm learning.",
        "Hello world. Hello tokenizer.",
        "12345 67890",
        "你好，世界！我在学习NLP。",
    ]
    tokenizer = BPETokenizer(vocab_size=512, special_tokens=["<|bos|>", "<|eos|>"])
    tokenizer.train(corpus)

    text = "<|bos|> Hello, world! 12345 <|eos|>"
    ids = tokenizer.encode(text)
    restored = tokenizer.decode(ids)

    logger.info("ids:", ids[:30], "...", f"(len={len(ids)})")
    logger.info("restored:", restored)

当你把 vocab_size 调大、语料覆盖更充分时，会观察到更常见的片段被逐渐合并（序列变短）；而低频片段仍会停留在较细粒度（最差也能退回到字节级，从而保证“永远可编码”）。

8. 特殊 token：为什么必须“原子化”

在聊天模板、控制标记、工具调用等场景里，特殊 token 承担“结构语义”，通常需要满足：

  • 精确匹配：不会被预分词拆开，也不会参与 BPE 合并。
  • 稳定映射：映射到固定 id，方便下游进行对齐、掩码与监督。

因此工程上一般会先把特殊 token 从文本中切出，剩余部分再进入“预分词 + BPE”的主流程。

9. 词表大小与数字分组：两个现实的 trade-off

9.1 词表大小（vocab_size）

词表大小决定了“序列长度”与“参数规模”的平衡：

  • 小词表：embedding 参数少、训练更省，但序列更长（推理更慢）。
  • 大词表：序列更短、推理更快，但 embedding 占比更高，也更容易为低频片段浪费容量。

实践中常见 2^N（如 32768、65536），主要是工程习惯与对齐便利，并非硬性规定。

9.2 数字分组（\p{N}{1,2} vs \p{N}{1,3}）

把数字切成更短的 chunk（例如 1~2 位）可以减少“数字组合”对词表的占用，但会让数字序列更长。对小词表/小模型，这往往是合理折中；对强依赖数学表达的任务，可能需要更谨慎的策略（或在相关语料上重新训练）。

10. BPB（bits per byte）：更公平地比较 tokenizer

仅比较 token-level loss 很容易被“token 粒度”误导：同一句话，如果 tokenizer 让 token 更长，token 数变少，token-level loss 可能看起来更低，但不代表模型真的更强。

一种更稳妥的归一化方式是把损失折算到字节级别。

设平均 token-level 交叉熵为 $H$（单位：nats/token），平均每个 token 覆盖 $B$ 字节（bytes/token），则：

$$ \mathrm{BPB} = \frac{H / \ln 2}{B} $$

BPB 的单位是 bits/byte，越小越好。这样不同 tokenizer、不同词表大小之间更可比。

11. 常见问题

1. 为什么同一词在不同上下文切分不同？

因为预分词通常保留前导空格，"Hello" 与 " Hello" 是不同 chunk；而 BPE 合并只在 chunk 内发生。

2. 为什么空格敏感？

这是一种把“词边界/位置信息”编码进 token 的设计：句首与句中的分布差异很大，显式区分有助于模型学习。

3. 中文会被切得更碎吗？

字节级 BPE 保证能编码中文，但如果训练语料中文占比低，常见汉字对应的字节组合不够“频繁”，就不容易被合并，导致 token 更多。

12. 实践建议

  • 想要复现某个模型的切分：预分词规则 + merges + special tokens 缺一不可。
  • 中文为主的模型：字节级并不等价于“中文友好”，通常仍需要中文语料参与训练（或选择更贴近中文的方案，如 SentencePiece）。
  • 做评测：跨 tokenizer 比较时，优先看 BPB 这类归一化指标，而不是裸 loss。

13. 总结

BPE 的核心并不复杂：训练阶段不断合并最频繁的相邻 token 对，得到有序 merges；推理阶段在 chunk 内按 rank 贪心合并。

当你把“正则预分词 + 前导空格 + 字节级基础词表 + 特殊 token 原子化”串起来，大多数 tokenizer 的“怪切分”都能得到一致解释，也更容易写出可复现、可部署的实现。

参考资料

  1. Philip Gage. A New Algorithm for Data Compression (1994).
  2. Rico Sennrich, Barry Haddow, Alexandra Birch. Neural Machine Translation of Rare Words with Subword Units (2016).
  3. Alec Radford et al. Language Models are Unsupervised Multitask Learners (2019)（GPT-2，byte-level BPE）。
  4. Taku Kudo, John Richardson. SentencePiece: A simple and language independent subword tokenizer and detokenizer for Neural Text Processing (2018).
  5. HuggingFace Tokenizers 文档。
  6. OpenAI tiktoken: OpenAI tiktoken GitHub
  7. nanochat: nanochat/tokenizer.py和rustbpe

致谢

  1. 感谢开源社区提供的代码实现与论文资料，为本文的理解与写作奠定了基础。
  2. 感谢 AI 工具在代码阅读（如 nanochat 等开源项目）及文章润色方面提供的协助。

本文力求准确，但疏漏难免。若发现内容或代码错误，欢迎反馈，以便勘误更新。^-^!