Tokenizer 往往藏在模型调用链的入口和出口,看起来只是文本与 token ID 之间的转换,却可能决定在线服务的首包延迟、批处理吞吐和 CPU 成本。围绕 Tokenizers v1,真正值得关注的不只是“更快”,而是编码、解码在不同输入规模和并发配置下如何变化,以及这些数字是否来自可复现的测量。
由于来源摘要没有给出具体硬件、数据集和性能数字,下面不复述未经说明的结论,而是给出一套可以直接落地的测试方法,用来验证目标 v1 构建在自己环境中的表现。
编码和解码不能混成一个数字
编码与解码走的是不同路径:
- 编码需要规范化文本、预分词、查找词表并应用分词算法。
- 解码从 token ID 查回 token,再处理空格、字节和特殊 token。
- 批量编码可能利用底层并行能力,单条短文本则更容易被调用开销支配。
- 长文本不仅增加计算量,也会放大临时对象和内存分配的成本。
因此,只报告“每秒处理多少条文本”通常不够。至少应同时记录:
- 每秒字符数或字节数;
- 每秒 token 数;
- 单批次耗时;
- p50、p95 等延迟分位数;
- 输入长度分布;
- 线程数、批大小、CPU 型号和库版本。
特别要避免只用固定短句测试。生产流量通常同时包含搜索词、聊天消息、代码片段和长文档,而这些输入对 tokenizer 的压力并不相同。
一份可复制的本地基准测试
下面的脚本会在本地生成语料、训练一个小型 Byte-level BPE tokenizer,然后分别测量批量编码和解码。它不依赖外部模型下载,适合先验证测试流程。要测 Tokenizers v1 的真实表现时,把脚本中的 tokenizer 构造部分替换为目标 v1 tokenizer,并固定其词表与配置。
先安装依赖:
python -m pip install tokenizers
python -m pip show tokenizers
保存为 bench_tokenizer.py:
import argparse
import random
import statistics
import tempfile
import time
from pathlib import Path
from tokenizers import Tokenizer
from tokenizers.decoders import ByteLevel as ByteLevelDecoder
from tokenizers.models import BPE
from tokenizers.pre_tokenizers import ByteLevel
from tokenizers.trainers import BpeTrainer
def make_texts(count: int) -> list[str]:
rng = random.Random(42)
words = [
"token", "model", "service", "latency", "throughput",
"python", "database", "缓存", "请求", "性能",
]
texts = []
for i in range(count):
length = rng.choice([8, 16, 32, 64])
body = " ".join(rng.choice(words) for _ in range(length))
texts.append(f"request={i} {body}")
return texts
def build_tokenizer() -> Tokenizer:
training_texts = make_texts(10_000)
with tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", delete=False
) as corpus:
corpus.write("\n".join(training_texts))
corpus_path = Path(corpus.name)
try:
tokenizer = Tokenizer(BPE(unk_token="<unk>"))
tokenizer.pre_tokenizer = ByteLevel(add_prefix_space=False)
tokenizer.decoder = ByteLevelDecoder()
trainer = BpeTrainer(
vocab_size=2_000,
min_frequency=2,
special_tokens=["<unk>"],
initial_alphabet=ByteLevel.alphabet(),
)
tokenizer.train([str(corpus_path)], trainer)
return tokenizer
finally:
corpus_path.unlink(missing_ok=True)
def measure(fn, rounds: int):
samples = []
result = None
for _ in range(rounds):
start = time.perf_counter()
result = fn()
samples.append(time.perf_counter() - start)
return statistics.median(samples), result
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--sizes", type=int, nargs="+", default=[100, 1_000, 10_000]
)
parser.add_argument("--rounds", type=int, default=7)
args = parser.parse_args()
tokenizer = build_tokenizer()
print(
"batch,chars,tokens,encode_ms,encode_tokens_s,"
"decode_ms,decode_tokens_s"
)
for size in args.sizes:
texts = make_texts(size)
# 预热,减少首次调用初始化对结果的干扰。
tokenizer.encode_batch(texts[: min(100, size)])
encode_seconds, encodings = measure(
lambda: tokenizer.encode_batch(texts), args.rounds
)
token_ids = [encoding.ids for encoding in encodings]
token_count = sum(len(ids) for ids in token_ids)
decode_seconds, decoded = measure(
lambda: tokenizer.decode_batch(token_ids), args.rounds
)
if decoded != texts:
raise RuntimeError("Round-trip check failed")
char_count = sum(len(text) for text in texts)
print(
f"{size},{char_count},{token_count},"
f"{encode_seconds * 1000:.2f},"
f"{token_count / encode_seconds:.0f},"
f"{decode_seconds * 1000:.2f},"
f"{token_count / decode_seconds:.0f}"
)
if __name__ == "__main__":
main()
运行测试:
python bench_tokenizer.py --sizes 100 1000 10000 50000 --rounds 7
输出是 CSV 风格的数据,可以直接重定向到文件:
python bench_tokenizer.py --sizes 100 1000 10000 50000 --rounds 11 \
> tokenizer-results.csv
这里同时执行了 round-trip 校验。性能优化如果破坏 decode(encode(text)) 的预期行为,就不能只看吞吐数字。实际模型可能会自动处理特殊 token、Unicode 规范化或前导空格,因此应按目标 tokenizer 的语义调整断言。
扩展性要看曲线,而不是单点峰值
所谓 scaling,至少有三个维度:
- 输入规模扩展:文本数量增加 10 倍时,耗时是否接近线性增长?
- 批大小扩展:从单条请求变成批量请求后,固定开销能否被摊薄?
- CPU 并行扩展:增加线程后,吞吐提升是否足以覆盖调度和内存带宽成本?
如果目标构建使用 Rayon 一类线程池,可以尝试用不同线程数运行同一组负载。以下环境变量是否生效取决于具体实现,执行前应查验目标版本的线程配置方式:
for threads in 1 2 4 8; do
echo "threads=${threads}"
TOKENIZERS_PARALLELISM=true \
RAYON_NUM_THREADS="${threads}" \
python bench_tokenizer.py --sizes 50000 --rounds 7
done
不要把“8 线程比 1 线程快”直接等同于良好扩展。可以计算:
加速比 = 单线程耗时 / N 线程耗时
并行效率 = 加速比 / N
例如,线程数继续增加但 tokens/s 不再增长,瓶颈可能已经转移到内存带宽、数据复制、Python 与原生代码边界,或者容器的 CPU 配额。在线服务还要警惕嵌套并行:Web Server 启动多个 worker,而每个 tokenizer worker 又创建一组线程,最终可能造成 CPU 过度订阅和尾延迟恶化。
让结果能用于生产决策
本地合成语料适合检查趋势,但不能代替生产数据。更可靠的测试集应该保留真实的长度分布、语言比例和内容类型,同时删除或脱敏敏感信息。建议至少拆分以下场景:
- 大量短输入,观察调用开销和 p95 延迟;
- 中等长度批处理,观察 tokens/s;
- 接近上下文上限的长文本,观察内存峰值;
- 中英文、emoji、组合字符和代码混合文本;
- 包含特殊 token、未知字符和异常 Unicode 的边界输入。
比较旧版本与 v1 时,还需要固定模型词表、配置、机器、电源模式、线程数和测试语料。每组测试先预热,再执行多轮并报告中位数及分位数,而不是挑选最好的一次。若 tokenizer 输出发生变化,则应先确认这是不是预期的语义变更,再讨论速度提升。
采用前的检查清单
Tokenizers v1 是否值得升级,最终应由完整链路的数据决定,而不是由一个孤立的微基准决定:
- 编码结果是否与当前模型和词表兼容?
- 解码后的特殊 token、空格和 Unicode 行为是否符合预期?
- 单请求延迟和批量吞吐是否都得到改善?
- 增加线程后,吞吐、CPU 占用和尾延迟如何变化?
- 容器 CPU 配额下是否仍能复现裸机结果?
- 新版本的初始化时间和常驻内存是否可接受?
最稳妥的做法是先建立正确性样本,再跑可复现的微基准,随后用生产流量回放验证端到端收益。Tokenizer 位于每次模型调用的必经之路,小幅改进可能被高调用量放大;同样,一个被忽视的兼容性差异也会被成倍放大。