大模型应用：大模型的词元化处理详解：BPE、WordPiece、Unigram.11

import re
from collections import defaultdict, Counter

class BPEChineseTokenizer:
    def __init__(self, vocab_size=10):
        self.vocab_size = vocab_size  # 目标词汇表大小
        self.vocab = {}  # 最终词汇表
        self.merge_rules = {}  # 合并规则（(a,b) → ab）
    
    def preprocess(self, corpus):
        """预处理：拆分为单字+结束符，统一格式"""
        processed = []
        for sentence in corpus:
            # 中文单字拆分，每个字后加</w>，词之间用空格分隔（这里按句子拆分）
            tokens = [char + '</w>' for char in sentence]
            processed.append(' '.join(tokens))
        return processed
    
    def get_pair_freq(self, corpus):
        """统计相邻字符对的频率"""
        pair_freq = defaultdict(int)
        for sentence in corpus:
            tokens = sentence.split()
            for i in range(len(tokens)-1):
                pair = (tokens[i], tokens[i+1])
                pair_freq[pair] += 1
        return pair_freq
    
    def merge_pair(self, corpus, pair, new_token):
        """合并语料中的指定字符对"""
        merged_corpus = []
        pattern = re.escape(f' {pair[0]} {pair[1]} ')
        replacement = f' {new_token} '
        for sentence in corpus:
            # 替换所有匹配的字符对
            merged_sentence = re.sub(pattern, replacement, f' {sentence} ').strip()
            merged_corpus.append(merged_sentence)
        return merged_corpus
    
    def train(self, corpus):
        """训练BPE分词器"""
        # 预处理语料
        processed_corpus = self.preprocess(corpus)
        # 初始化词汇表：所有单字
        all_tokens = []
        for sentence in processed_corpus:
            all_tokens.extend(sentence.split())
        initial_vocab = list(set(all_tokens))
        self.vocab = {token: idx for idx, token in enumerate(initial_vocab)}
        
        # 迭代合并直到达到词汇表大小
        while len(self.vocab) < self.vocab_size:
            # 统计字符对频率
            pair_freq = self.get_pair_freq(processed_corpus)
            if not pair_freq:
                break  # 无可用合并对
            # 找频率最高的对
            best_pair = max(pair_freq, key=pair_freq.get)
            # 生成新Token
            new_token = ''.join(best_pair).replace('</w>', '') + '</w>'
            # 记录合并规则
            self.merge_rules[best_pair] = new_token
            # 合并语料中的该对
            processed_corpus = self.merge_pair(processed_corpus, best_pair, new_token)
            # 更新词汇表
            if new_token not in self.vocab:
                self.vocab[new_token] = len(self.vocab)
        
        print("BPE训练完成！")
        print("合并规则：", self.merge_rules)
        print("最终词汇表：", self.vocab)
    
    def tokenize(self, text):
        """对新文本分词"""
        # 预处理文本为单字
        tokens = [char + '</w>' for char in text]
        # 应用合并规则（从长到短匹配）
        # 先将合并规则按新Token长度降序排序
        sorted_merges = sorted(self.merge_rules.items(), 
                              key=lambda x: len(x[1]), reverse=True)
        # 迭代合并
        while True:
            merged = False
            for (pair, new_token) in sorted_merges:
                if pair[0] in tokens and pair[1] in tokens:
                    # 找到相邻的pair
                    idx = tokens.index(pair[0])
                    if idx + 1 < len(tokens) and tokens[idx+1] == pair[1]:
                        # 合并
                        tokens = tokens[:idx] + [new_token] + tokens[idx+2:]
                        merged = True
                        break
            if not merged:
                break
        # 转换为词汇表ID
        token_ids = [self.vocab.get(token, -1) for token in tokens]
        return tokens, token_ids

# 测试代码
if __name__ == "__main__":
    # 中文语料
    corpus = ["我爱中国", "中国很强大", "我爱北京", "北京是中国首都"]
    # 初始化并训练BPE分词器
    bpe_tokenizer = BPEChineseTokenizer(vocab_size=10)
    bpe_tokenizer.train(corpus)
    # 分词测试
    test_text = "我爱北京"
    tokens, token_ids = bpe_tokenizer.tokenize(test_text)
    print(f"\n测试文本：{test_text}")
    print(f"分词结果：{tokens}")
    print(f"Token ID：{token_ids}")

import re
import math
from collections import defaultdict, Counter

class WordPieceChineseTokenizer:
    def __init__(self, vocab_size=10):
        self.vocab_size = vocab_size
        self.vocab = {}
        self.merge_rules = {}
        self.token_freq = defaultdict(int)  # Token频率
    
    def preprocess(self, corpus):
        """预处理：拆分为单字+结束符"""
        processed = []
        total_tokens = 0
        for sentence in corpus:
            tokens = [char + '</w>' for char in sentence]
            processed.append(' '.join(tokens))
            # 统计初始Token频率
            for token in tokens:
                self.token_freq[token] += 1
            total_tokens += len(tokens)
        self.total_tokens = total_tokens  # 总Token数
        return processed
    
    def calculate_gain(self, a, b, corpus):
        """计算合并(a,b)的对数似然增益"""
        # 统计a、b、ab的频率
        a_freq = self.token_freq.get(a, 0)
        b_freq = self.token_freq.get(b, 0)
        # 统计ab的共现频率
        ab_freq = 0
        for sentence in corpus:
            tokens = sentence.split()
            for i in range(len(tokens)-1):
                if tokens[i] == a and tokens[i+1] == b:
                    ab_freq += 1
        if a_freq == 0 or b_freq == 0 or ab_freq == 0:
            return -float('inf')  # 无增益
        
        # 计算概率
        p_a = a_freq / self.total_tokens
        p_b = b_freq / self.total_tokens
        p_ab = ab_freq / (self.total_tokens - ab_freq)  # 合并后总Token数减少ab_freq
        
        # 对数似然增益
        gain = math.log(p_ab) - math.log(p_a) - math.log(p_b)
        return gain
    
    def merge_pair(self, corpus, pair, new_token):
        """合并语料中的指定字符对"""
        merged_corpus = []
        pattern = re.escape(f' {pair[0]} {pair[1]} ')
        replacement = f' {new_token} '
        for sentence in corpus:
            merged_sentence = re.sub(pattern, replacement, f' {sentence} ').strip()
            merged_corpus.append(merged_sentence)
        # 更新Token频率
        a, b = pair
        ab_freq = self.token_freq.get(a, 0) + self.token_freq.get(b, 0) - (self.token_freq.get(new_token, 0))
        self.token_freq[new_token] = ab_freq
        del self.token_freq[a]
        del self.token_freq[b]
        self.total_tokens -= ab_freq  # 总Token数减少
        return merged_corpus
    
    def train(self, corpus):
        """训练WordPiece分词器"""
        processed_corpus = self.preprocess(corpus)
        # 初始化词汇表
        initial_vocab = list(self.token_freq.keys())
        self.vocab = {token: idx for idx, token in enumerate(initial_vocab)}
        
        while len(self.vocab) < self.vocab_size:
            # 生成所有候选字符对
            candidate_pairs = set()
            for sentence in processed_corpus:
                tokens = sentence.split()
                for i in range(len(tokens)-1):
                    candidate_pairs.add((tokens[i], tokens[i+1]))
            if not candidate_pairs:
                break
            
            # 计算每个候选对的增益
            gains = {}
            for pair in candidate_pairs:
                gain = self.calculate_gain(pair[0], pair[1], processed_corpus)
                gains[pair] = gain
            
            # 找增益最高的对
            best_pair = max(gains, key=gains.get)
            best_gain = gains[best_pair]
            if best_gain <= 0:
                break  # 增益≤0，停止合并
            
            # 生成新Token
            new_token = ''.join(best_pair).replace('</w>', '') + '</w>'
            self.merge_rules[best_pair] = new_token
            # 合并语料
            processed_corpus = self.merge_pair(processed_corpus, best_pair, new_token)
            # 更新词汇表
            if new_token not in self.vocab:
                self.vocab[new_token] = len(self.vocab)
        
        print("WordPiece训练完成！")
        print("合并规则：", self.merge_rules)
        print("最终词汇表：", self.vocab)
    
    def tokenize(self, text):
        """分词（最长匹配）"""
        tokens = [char + '</w>' for char in text]
        # 按新Token长度降序应用合并规则
        sorted_merges = sorted(self.merge_rules.items(), 
                              key=lambda x: len(x[1]), reverse=True)
        while True:
            merged = False
            for (pair, new_token) in sorted_merges:
                if pair[0] in tokens and pair[1] in tokens:
                    idx = tokens.index(pair[0])
                    if idx + 1 < len(tokens) and tokens[idx+1] == pair[1]:
                        tokens = tokens[:idx] + [new_token] + tokens[idx+2:]
                        merged = True
                        break
            if not merged:
                break
        token_ids = [self.vocab.get(token, -1) for token in tokens]
        return tokens, token_ids

# 测试代码
if __name__ == "__main__":
    corpus = ["我爱中国", "中国很强大", "我爱北京", "北京是中国首都"]
    wp_tokenizer = WordPieceChineseTokenizer(vocab_size=10)
    wp_tokenizer.train(corpus)
    # 分词测试
    test_text = "中国很强大"
    tokens, token_ids = wp_tokenizer.tokenize(test_text)
    print(f"\n测试文本：{test_text}")
    print(f"分词结果：{tokens}")
    print(f"Token ID：{token_ids}")

import math
from collections import defaultdict, Counter
from itertools import combinations

class UnigramChineseTokenizer:
    def __init__(self, vocab_size=10):
        self.vocab_size = vocab_size
        self.vocab = {}
        self.token_prob = defaultdict(float)  # Token概率
    
    def generate_candidates(self, corpus, max_len=3):
        """生成候选Token（单字、双字、三字）"""
        candidates = set()
        for sentence in corpus:
            # 生成所有长度≤max_len的连续子串
            for i in range(len(sentence)):
                for j in range(1, min(max_len+1, len(sentence)-i+1)):
                    token = sentence[i:i+j]
                    candidates.add(token)
        return list(candidates)
    
    def calculate_token_freq(self, corpus, candidates):
        """统计候选Token的频率"""
        freq = defaultdict(int)
        total = 0
        for sentence in corpus:
            # 统计每个候选Token的出现次数
            for token in candidates:
                token_len = len(token)
                for i in range(len(sentence)-token_len+1):
                    if sentence[i:i+token_len] == token:
                        freq[token] += 1
                        total += 1
        return freq, total
    
    def calculate_perplexity(self, corpus, vocab, token_prob):
        """计算语料的困惑度"""
        total_log_prob = 0
        total_tokens = 0
        for sentence in corpus:
            # 找到最优切分（概率最高）
            best_log_prob = -float('inf')
            # 简单实现：最长匹配找切分
            tokens = self._longest_match(sentence, vocab)
            # 计算该切分的对数概率
            log_prob = sum([math.log(token_prob.get(t, 1e-10)) for t in tokens])
            total_log_prob += log_prob
            total_tokens += len(tokens)
        # 困惑度 = exp(-平均对数概率)
        avg_log_prob = total_log_prob / total_tokens
        perplexity = math.exp(-avg_log_prob)
        return perplexity
    
    def _longest_match(self, text, vocab):
        """最长匹配切分（用于简化困惑度计算）"""
        tokens = []
        i = 0
        vocab_sorted = sorted(vocab, key=len, reverse=True)  # 按长度降序
        while i < len(text):
            matched = False
            for token in vocab_sorted:
                token_len = len(token)
                if i + token_len <= len(text) and text[i:i+token_len] == token:
                    tokens.append(token)
                    i += token_len
                    matched = True
                    break
            if not matched:
                # 匹配失败，取单字
                tokens.append(text[i])
                i += 1
        return tokens
    
    def train(self, corpus):
        """训练Unigram分词器"""
        # 步骤1：生成候选Token
        candidates = self.generate_candidates(corpus, max_len=3)
        # 步骤2：统计频率，初始化概率
        freq, total = self.calculate_token_freq(corpus, candidates)
        # 初始化Vocab（过滤掉频率为0的Token）
        initial_vocab = [t for t in candidates if freq[t] > 0]
        current_vocab = initial_vocab.copy()
        
        # 步骤3：迭代删除Token直到达到目标大小
        while len(current_vocab) > self.vocab_size:
            # 计算当前Token概率
            current_freq, current_total = self.calculate_token_freq(corpus, current_vocab)
            current_prob = {t: current_freq[t]/current_total for t in current_vocab}
            # 计算当前困惑度
            current_pp = self.calculate_perplexity(corpus, current_vocab, current_prob)
            
            # 计算删除每个Token后的困惑度
            pp_dict = {}
            for token in current_vocab:
                # 临时删除该Token
                temp_vocab = [t for t in current_vocab if t != token]
                if not temp_vocab:
                    pp_dict[token] = float('inf')
                    continue
                # 计算临时概率
                temp_freq, temp_total = self.calculate_token_freq(corpus, temp_vocab)
                temp_prob = {t: temp_freq[t]/temp_total for t in temp_vocab}
                # 计算临时困惑度
                temp_pp = self.calculate_perplexity(corpus, temp_vocab, temp_prob)
                pp_dict[token] = temp_pp
            
            # 找到困惑度上升最小的Token（即pp_dict最小的）
            best_token_to_remove = min(pp_dict, key=pp_dict.get)
            current_vocab.remove(best_token_to_remove)
        
        # 最终Vocab和概率
        final_freq, final_total = self.calculate_token_freq(corpus, current_vocab)
        self.token_prob = {t: final_freq[t]/final_total for t in current_vocab}
        self.vocab = {t: idx for idx, t in enumerate(current_vocab)}
        
        print("Unigram训练完成！")
        print("最终词汇表：", self.vocab)
        print("Token概率：", self.token_prob)
    
    def tokenize(self, text):
        """最优概率切分"""
        # 动态规划找最优切分
        n = len(text)
        # dp[i]：前i个字符的最大对数概率
        dp = [-float('inf')] * (n+1)
        dp[0] = 0.0
        # prev[i]：前i个字符的最优切分位置
        prev = [0] * (n+1)
        
        for i in range(1, n+1):
            for j in range(max(0, i-3), i):  # 最多匹配3字
                token = text[j:i]
                if token in self.token_prob:
                    log_prob = math.log(self.token_prob[token])
                    if dp[j] + log_prob > dp[i]:
                        dp[i] = dp[j] + log_prob
                        prev[i] = j
        
        # 回溯找切分结果
        tokens = []
        i = n
        while i > 0:
            j = prev[i]
            tokens.append(text[j:i])
            i = j
        tokens = tokens[::-1]
        token_ids = [self.vocab.get(t, -1) for t in tokens]
        return tokens, token_ids

# 测试代码
if __name__ == "__main__":
    corpus = ["我爱中国", "中国很强大", "我爱北京", "北京是中国首都"]
    unigram_tokenizer = UnigramChineseTokenizer(vocab_size=10)
    unigram_tokenizer.train(corpus)
    # 分词测试
    test_text = "北京是中国首都"
    tokens, token_ids = unigram_tokenizer.tokenize(test_text)
    print(f"\n测试文本：{test_text}")
    print(f"分词结果：{tokens}")
    print(f"Token ID：{token_ids}")