每日基础课

【生活类比：乐高拼装说明书】 把一句话或者一段带说话人标签的语音，想象成一只组装好的乐高恐龙。 神经网络像一台严格按零件编号运作的机械臂，它无法直接理解“整只恐龙”。 Tokenization 的工作就是把恐龙拆成标准积木块（Tokens），并翻开零件清单（词表 Vocabulary），在每块积木上刻上零件号（Token ID）。机械臂拿到这串数字，才能开始计算。

【必要公式与符号解释】 给定原始文本序列 $X$： $$X = (c_1, c_2, \dots, c_T) \xrightarrow{\text{Tokenizer}} T = (i_1, i_2, \dots, i_N)$$ - $c_t$：原始输入中的最小基础单元（如单个字符或字节）。 - $N$：分词后的 Token 数量（通常 $N \le T$）。 - $i_n$：第 $n$ 个积木的编号（Token ID），且 $i_n \in [0, |V|-1]$。 - $|V|$：词表大小（Vocabulary Size），即模型认识的所有积木总数。

【它解决什么问题？没有它会怎样？】 计算机本质上只能做矩阵乘法，看不懂人类字符。 1. 没有它，按整词切分：遇到新词、人名或口语化变形（如“Awwwwsome”）直接崩溃，词表膨胀到几百万，显存直接撑爆（OOV 问题，Out of Vocabulary）。 2. 没有它，按纯字符/拼音切分：一个句子被切成上千个字母，序列太长，Transformer 的自注意力计算量呈平方级爆炸（$O(N^2)$），长音频转录直接跑不动。 Tokenization（特别是子词算法）在“序列长度”与“词表大小”之间找到了最佳平衡点。

【在主流模型与音视频/说话人日志中的实际位置】 它是模型的最前线海关： 1. 文本与 ASR：在 Whisper 或转录模型中，文本提示和预测文本均先由 Tokenizer 转为 ID。 2. 说话人日志（Diarization + ASR）：在端到端联合任务中，特殊标签如 `<spk1>`、`<spk2>` 会作为保留 Token 写入词表，和识别出的字词一起排成序列输出。 3. 学习指引：想建立模型输入端的数据流直觉，推荐阅读 *The Illustrated Transformer*；想在 PyTorch 代码里定位 Tokenizer 索引与 Embedding 的衔接点，看 *The Annotated Transformer*。

【解题思路】 1. 定义特殊标记：`<pad>`（填充）、`<unk>`（未知词）、`<spk1>`、`<spk2>`（说话人标记）。 2. 构建映射字典：`token2id` 和 `id2token`。 3. 编写 `encode` 与 `decode` 函数。 4. 编写 `batch_encode` 处理长短不一的句子，填充 `<pad>` 并生成注意力掩码。

import torch

class SimpleAudioDiarizationTokenizer:
    def __init__(self, special_tokens=None):
        if special_tokens is None:
            special_tokens = ["<pad>", "<unk>", "<spk1>", "<spk2>"]
        self.special_tokens = special_tokens
        self.pad_token = "<pad>"
        self.unk_token = "<unk>"
        
        self.token2id = {}
        self.id2token = {}
        
        # 优先填入特殊 token
        for token in self.special_tokens:
            self._add_token(token)

    def _add_token(self, token):
        if token not in self.token2id:
            new_id = len(self.token2id)
            self.token2id[token] = new_id
            self.id2token[new_id] = token
            return new_id
        return self.token2id[token]

    def build_vocab(self, corpus):
        """根据语料库建立字符级词表（保留特殊 token）"""
        for text in corpus:
            tokens = self._tokenize_text(text)
            for t in tokens:
                self._add_token(t)
        print(f"词表构建完成，词表大小: {len(self.token2id)}")

    def _tokenize_text(self, text):
        """简易切词：识别特殊标记，其余按字符切分"""
        tokens = []
        i = 0
        while i < len(text):
            # 检查是否匹配特殊标记
            matched = False
            for spk in self.special_tokens:
                if text[i:].startswith(spk):
                    tokens.append(spk)
                    i += len(spk)
                    matched = True
                    break
            if not matched:
                if text[i] != " ":  # 忽略空格以简化
                    tokens.append(text[i])
                i += 1
        return tokens

    def encode(self, text):
        tokens = self._tokenize_text(text)
        unk_id = self.token2id[self.unk_token]
        return [self.token2id.get(t, unk_id) for t in tokens]

    def decode(self, ids):
        return "".join([self.id2token.get(i, self.unk_token) for i in ids if self.id2token.get(i) != self.pad_token])

    def batch_encode(self, text_list):
        """批量编码并对齐 Padding"""
        encoded_batch = [self.encode(text) for text in text_list]
        max_len = max(len(seq) for seq in encoded_batch)
        pad_id = self.token2id[self.pad_token]
        
        input_ids = []
        attention_mask = []
        
        for seq in encoded_batch:
            pad_len = max_len - len(seq)
            padded_seq = seq + [pad_id] * pad_len
            mask = [1] * len(seq) + [0] * pad_len
            input_ids.append(padded_seq)
            attention_mask.append(mask)
            
        return torch.tensor(input_ids, dtype=torch.long), torch.tensor(attention_mask, dtype=torch.long)

# ================= 测试运行 =================
if __name__ == "__main__":
    # 模拟包含说话人日志的转录数据
    corpus = [
        "<spk1> 你好，听得到吗？",
        "<spk2> 听得到，我们开始会议吧。",
        "<spk1> 好的，今天讨论模型分词。"
    ]
    
    tokenizer = SimpleAudioDiarizationTokenizer()
    tokenizer.build_vocab(corpus)
    
    # 编码测试
    test_batch = [
        "<spk1> 你好",
        "<spk2> 今天讨论模型分词。"
    ]
    
    input_ids, attention_mask = tokenizer.batch_encode(test_batch)
    
    print("\n--- 编码结果 ---")
    print("Input IDs Tensor 形状:", input_ids.shape)
    print("Input IDs:\n", input_ids)
    print("Attention Mask:\n", attention_mask)
    
    print("\n--- 解码验证 ---")
    for i, seq_ids in enumerate(input_ids.tolist()):
        print(f"样本 {i+1} 解码:", tokenizer.decode(seq_ids))