Tokenizers v1 性能实测:编码、解码与扩展性该怎么量

2026-09-21 14 预计阅读时间: 1 分钟
来源: huggingface.co AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

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,至少有三个维度:

  1. 输入规模扩展:文本数量增加 10 倍时,耗时是否接近线性增长?
  2. 批大小扩展:从单条请求变成批量请求后,固定开销能否被摊薄?
  3. 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 位于每次模型调用的必经之路,小幅改进可能被高调用量放大;同样,一个被忽视的兼容性差异也会被成倍放大。


相关推荐