1. 什么是分词器? #

2. 前置知识 #

2.1 什么是 Token? #

2.2 Token 与 API 计费 #

2.3 用哪个分词器? #

API 平台 分词库 典型模型
OpenAI tiktoken gpt-4o、gpt-3.5-turbo
DeepSeek deepseek_tokenizer deepseek-v4-pro、deepseek-v4-flash

3. tiktoken(OpenAI 系列) #

3.1 安装 #

# 说明:安装 tiktoken 分词库
py -m pip install tiktoken

3.2 验证安装 #

# -*- coding: utf-8 -*-
# 说明:导入 tiktoken 库
import tiktoken

# 说明:打印 tiktoken 版本号
print("tiktoken 版本:", tiktoken.__version__)

# 说明:按模型名获取编码器(gpt-4o 使用 o200k_base)
enc = tiktoken.encoding_for_model("gpt-4o")

# 说明:将文本编码为 Token 列表
tokens = enc.encode("hello world")

# 说明:打印 Token 数量
print("Token 数量:", len(tokens))

# 说明:打印验证通过提示
print("验证通过")

3.3 基本用法 #

# -*- coding: utf-8 -*-
# 说明:导入 tiktoken
import tiktoken

# 说明:按目标模型名获取编码器(与 API 调用时使用同一模型名)
enc = tiktoken.encoding_for_model("gpt-4o")

# 说明:待统计的文本
text = "Python 是一门简洁易学的编程语言。"

# 说明:编码为 Token 列表
tokens = enc.encode(text)

# 说明:打印 Token 数量(项目中最常用的写法)
print("Token 数量:", len(tokens))

# 说明:解码还原原文(验证 encode 是否正确)
print("解码结果:", enc.decode(tokens))

3.4 模型与编码对照 #

编码名称 典型模型
o200k_base GPT-4o、GPT-4o-mini、GPT-4.1、o1、o3-mini
cl100k_base GPT-4、GPT-3.5-turbo、text-embedding-3-small

3.5 统计单段文本 #

# -*- coding: utf-8 -*-
# 说明:导入 tiktoken
import tiktoken


def count_tokens(text: str, model: str = "gpt-4o") -> int:
    """统计一段文本在 OpenAI 模型下的 Token 数量。"""
    # 说明:按模型名获取编码器
    enc = tiktoken.encoding_for_model(model)
    # 说明:编码后返回 Token 个数
    return len(enc.encode(text))


# 说明:测试 Prompt 文本
prompt = "请用三句话介绍 Python 的特点。"

# 说明:统计 Token 数并打印
print("Token 数量:", count_tokens(prompt))

3.6 统计 Chat API 消息列表 #

# -*- coding: utf-8 -*-
# 说明:导入 tiktoken
import tiktoken


def count_message_tokens(messages: list, model: str = "gpt-4o") -> int:
    """估算 OpenAI Chat API messages 列表的总 Token 数(近似值)。"""
    # 说明:按模型获取编码器
    enc = tiktoken.encoding_for_model(model)
    # 说明:每条消息的固定格式开销
    tokens_per_message = 3
    # 说明:累计 Token 总数
    total = 0
    # 说明:遍历每条消息
    for msg in messages:
        # 说明:加上本条消息的固定开销
        total += tokens_per_message
        # 说明:遍历 role、content 等字段并累加 Token
        for key, value in msg.items():
            total += len(enc.encode(value))
    # 说明:加上模型回复的起始开销
    total += 3
    return total


# 说明:构造典型的对话消息
messages = [
    {"role": "system", "content": "你是一名 Python 编程助手,回答简洁清晰。"},
    {"role": "user", "content": "列表和元组有什么区别?"},
]

# 说明:估算并打印总 Token 数
print("messages 约", count_message_tokens(messages), "tokens")

4. deepseek_tokenizer(DeepSeek 系列) #

4.1 安装 #

# 说明:安装 DeepSeek 官方分词库
py -m pip install deepseek-tokenizer

4.2 基本用法 #

# -*- coding: utf-8 -*-
# 说明:从 deepseek_tokenizer 导入官方分词器
from deepseek_tokenizer import ds_token

# 说明:待统计的文本
text = "你是谁?"

# 说明:编码为 Token 列表
tokens = ds_token.encode(text)

# 说明:打印 Token 列表
print("Token 列表:", tokens)

# 说明:打印纯文本 Token 数量
print("内容 Token 数:", len(tokens))

# 说明:解码还原原文
print("解码结果:", ds_token.decode(tokens))

4.3 封装计数函数 #

# -*- coding: utf-8 -*-
# 说明:导入 DeepSeek 官方分词器
from deepseek_tokenizer import ds_token


def count_deepseek_tokens(text: str) -> int:
    """统计 DeepSeek 模型下一段纯文本的 Token 数量。"""
    # 说明:编码后返回 Token 个数
    return len(ds_token.encode(text))


# 说明:测试 Prompt 文本
prompt = "请用三句话介绍 Python 的特点。"

# 说明:统计并打印 Token 数
print("DeepSeek 内容 Token 数:", count_deepseek_tokens(prompt))

4.4 两种分词器对比 #

# -*- coding: utf-8 -*-
# 说明:导入 tiktoken
import tiktoken

# 说明:导入 DeepSeek 官方分词器
from deepseek_tokenizer import ds_token

# 说明:对比用的短文本
text = "你是谁?"

# 说明:用 tiktoken(OpenAI 编码,不能用于 DeepSeek 计费)
tiktoken_count = len(tiktoken.get_encoding("cl100k_base").encode(text))

# 说明:用 DeepSeek 官方分词器
deepseek_count = len(ds_token.encode(text))

# 说明:打印对比结果
print(f"文本:{text!r}")
print(f"tiktoken (cl100k_base):{tiktoken_count} tokens  ← OpenAI 用")
print(f"deepseek_tokenizer:    {deepseek_count} tokens  ← DeepSeek 用")
print("Chat API prompt_tokens 还会再加消息格式开销,以 usage 为准")

5. BPE 分词原理 #

5.1 BPE 的两个阶段 #

5.2 训练与编码时序 #

sequenceDiagram participant 用户 participant BPE as SimpleBPE participant 规则 as 词表与合并规则 Note over 用户,规则: 阶段一:训练(离线,只需一次) 用户->>BPE: train(语料, num_merges) loop 重复 num_merges 次 BPE->>BPE: 统计所有相邻字符对出现次数 BPE->>BPE: 找出频率最高的字符对(如 h + e) BPE->>规则: 合并为新 Token,写入 merges end BPE-->>用户: 训练完成 Note over 用户,规则: 阶段二:编码(每次输入文本时) 用户->>BPE: encode("hello world") loop 按 merges 顺序逐轮合并 BPE->>BPE: 将匹配的相邻字符对替换为新 Token end BPE->>规则: 查词表,转为 Token ID BPE-->>用户: 返回 Token ID 列表

5.3 编码单词的合并过程 #

sequenceDiagram participant 用户 participant BPE as SimpleBPE participant 序列 as token 序列 用户->>BPE: encode("hello") BPE->>序列: 初始 ['h','e','l','l','o'] BPE->>序列: 合并 h+e -> ['he','l','l','o'] BPE->>序列: 合并 he+l -> ['hel','l','o'] BPE->>序列: 合并 hel+l -> ['hell','o'] BPE->>序列: 合并 hell+o -> ['hello'] BPE->>BPE: 查词表 hello -> ID 11 BPE-->>用户: 返回 [11, ...]

5.4 BPE 实现 #

# -*- coding: utf-8 -*-
# 说明:从 collections 导入 defaultdict,用于统计字符对频率
from collections import defaultdict


# 定义SimpleBPE类
class SimpleBPE:
    # 构造函数
    def __init__(self):
        # 初始化词表,为字典类型(字符到id的映射)
        self.vocab = {}
        # 初始化合并规则,存储相邻字符对到新token的映射
        self.merges = {}
        # 初始化反向映射,token_id到token文本
        self.id_to_token = {}
        # 定义特殊结束标记
        self.end_token = "<|endoftext|>"

    # 训练BPE分词器的方法,参数为文本和合并次数
    def train(self, text, num_merges=100):
        """
        训练BPE分词器:学习合并频率最高的相邻字符对
        """
        print(f"\n【训练开始】语料: {text!r},计划合并 {num_merges} 轮")

        # 从文本中提取所有唯一字符并排序(排除空格),作为初始字符集
        chars = sorted(c for c in set(text) if c != " ")

        # 初始化词表,将所有字符映射为id,然后加上特殊结束标记
        self.vocab = {i: char for i, char in enumerate(chars)}
        # 为特殊结束标记分配一个id
        self.vocab[len(self.vocab)] = self.end_token
        # 下一个可用的token id
        next_id = len(self.vocab)

        print(f"初始词表 (共 {len(self.vocab)} 项):")
        for token_id, token_str in self.vocab.items():
            print(f"  ID {token_id:>2} -> {token_str!r}")

        # 将每个单词拆成字符,作为 BPE 训练的初始 token 序列
        word_tokens = [self._tokenize_word(word) for word in text.split()]

        # 进行多次合并迭代
        merge_round = 0
        for _ in range(num_merges):
            # 统计所有词的相邻字符对出现频率
            pair_counts = defaultdict(int)
            # 遍历所有token序列
            for token_list in word_tokens:
                # 遍历当前token序列中的所有相邻对
                for i in range(len(token_list) - 1):
                    # 取相邻两个token组成pair
                    pair = (token_list[i], token_list[i + 1])
                    # 该pair对应频次加1
                    pair_counts[pair] += 1

            # 如果没有可统计的pair,提前结束
            if not pair_counts:
                print(f"【合并轮次 {merge_round + 1}】无可合并的字符对,训练提前结束")
                break

            merge_round += 1

            # 找出出现频率最高的pair
            best_pair = max(pair_counts, key=pair_counts.get)
            best_count = pair_counts[best_pair]

            # 生成一个新token,为best_pair两个字符串拼接
            new_token = best_pair[0] + best_pair[1]
            # 新token分配新id,并写入vocab
            self.vocab[next_id] = new_token
            # 新合并规则写入merges
            self.merges[best_pair] = next_id

            print(
                f"【合并轮次 {merge_round}】"
                f" {best_pair[0]!r}+{best_pair[1]!r} -> {new_token!r}"
                f" (ID: {next_id}, 频次: {best_count})"
            )

            # 用新token替换word_tokens中所有best_pair出现的位置
            new_word_tokens = []
            for token_list in word_tokens:
                # 新的token序列
                new_list = []
                # 遍历token_list中的元素
                i = 0
                while i < len(token_list):
                    # 如果当前位置和下一个正好是best_pair
                    if i < len(token_list) - 1 and (token_list[i], token_list[i + 1]) == best_pair:
                        # 合并到新token
                        new_list.append(new_token)
                        # 跳过合并的两个位置
                        i += 2
                    else:
                        # 否则,正常加入该token
                        new_list.append(token_list[i])
                        i += 1
                new_word_tokens.append(new_list)
            # 更新word_tokens为本轮后的结果
            word_tokens = new_word_tokens

            # 新的id自增
            next_id += 1

        # 训练结束后,构建反向映射:token文本->token_id
        self.id_to_token = {v: k for k, v in self.vocab.items()}

        print(f"\n【训练结束】实际合并 {len(self.merges)} 轮,最终 token 序列: {word_tokens}")
        print(f"最终词表 (共 {len(self.vocab)} 项):")
        for token_id in sorted(self.vocab.keys()):
            print(f"  ID {token_id:>2} -> {self.vocab[token_id]!r}")

    # 编码函数,将文本转为token id序列
    def encode(self, text):
        """
        编码:将文本转换为token IDs
        """
        print(f"\n【编码】文本: {text!r}")

        # 将文本按照空格分割为单词
        words = text.split()
        # 结果token id列表
        result = []

        # 遍历所有单词
        for word in words:
            # 把单词拆为字符序列
            token_list = list(word)

            # 应用所有合并规则直到不能再合并
            changed = True
            while changed:
                # 先假定本轮没有变化
                changed = False
                i = 0
                # 新的token序列
                new_list = []
                # 遍历当前token_list
                while i < len(token_list):
                    # 如果当前和下一个字符组成的pair在合并规则中
                    if i < len(token_list) - 1:
                        pair = (token_list[i], token_list[i + 1])
                        if pair in self.merges:
                            merged_token = self.vocab[self.merges[pair]]
                            # 用合并生成的新token替换
                            new_list.append(merged_token)
                            # 跳过两个字符
                            i += 2
                            # 本次有合并,改为True
                            changed = True
                            # 跳到下一轮
                            continue
                    # 否则,正常加入当前字符
                    new_list.append(token_list[i])
                    i += 1
                # 更新token_list为本轮合并后的结果
                token_list = new_list

            # 把最终token序列转为id
            for token in token_list:
                # 遍历词表,找到token对应的id
                for token_id, token_str in self.vocab.items():
                    if token_str == token:
                        result.append(token_id)
                        break

        # 加入结束标记id
        end_id = self.id_to_token.get(self.end_token, -1)
        result.append(end_id)

        print(f"Token IDs: {result}")

        # 返回token id列表
        return result

    # 解码函数,将token id列表还原为字符串
    def decode(self, token_ids):
        """
        解码:将token IDs还原为文本
        """
        print(f"\n【解码】输入 IDs: {token_ids}")

        # 用于存放解码后的token字符串
        tokens = []
        # 遍历输入的token id
        for token_id in token_ids:
            # 保证id在词表中
            if token_id in self.vocab:
                # 获取对应token文本
                token_str = self.vocab[token_id]
                # 跳过特殊结束标记
                if token_str != self.end_token:
                    tokens.append(token_str)

        # 拼接返回最终字符串
        decoded = "".join(tokens)
        print(f"解码结果: {decoded!r}")
        return decoded

    # 把单词拆分成字符数组的辅助方法
    def _tokenize_word(self, word):
        """将单词拆分为字符"""
        # 直接按字符转为列表
        return list(word)


# ===================== 使用示例 =====================

# 定义demo方法进行功能演示
def main():
    bpe = SimpleBPE()

    corpus = "hello world hello hello world"
    bpe.train(corpus, num_merges=50)

    text = "hello world"
    token_ids = bpe.encode(text)

    tokens = [
        bpe.vocab[token_id]
        for token_id in token_ids
        if token_id in bpe.vocab and bpe.vocab[token_id] != "<|endoftext|>"
    ]
    decoded = bpe.decode(token_ids)

    print("\n【汇总结果】")
    print(f"文本: {text!r}")
    print(f"Token IDs: {token_ids}")
    print(f"Token 字符串: {tokens}")
    print(f"解码后: {decoded!r}")
    print(f"\n全部 {len(bpe.merges)} 条合并规则:")
    for (left, right), token_id in bpe.merges.items():
        print(f"  {left!r} + {right!r} -> {bpe.vocab[token_id]!r} (ID: {token_id})")


# 说明:判断是否以主程序方式运行
if __name__ == "__main__":
    # 说明:调用演示函数
    main()

6. 常见问题 #

6.1 编码结果和示例不一致 #

6.2 中文为什么更耗 Token? #

6.3 本地计数与 API usage 有差异 #

6.4 模型名找不到 #