五笔BPE分词器的技术演进与关键修复

在中文自然语言处理任务中，五笔编码与Byte Pair Encoding (BPE)的结合提供了一种独特的方法。本文分享在构建五笔BPE分词器过程中遇到的技术挑战及其解决方案。

问题背景

五笔BPE分词器旨在：

但在实现过程中，发现了两个关键问题：

问题一：临时文件权限冲突

症状：在Windows系统上训练后，临时文件无法正常删除，出现权限错误：

PermissionError: [WinError 32] The process cannot access the file because it is being used by another process

根源：

问题二：解码逻辑缺陷

症状：对混合内容解码时出现错误：

"你好abc" → 解码为 "你bca好" (非中文字符位置错乱)

根源：

修复方案

修复一：健壮的临时文件管理

def train(self, corpus_path: str, vocab_size: int, chunk_size: int = 10000):
# 生成唯一文件路径
timestamp = str(int(time.time()))
temp_path = os.path.join("temp", f"wubi_bpe_temp_{timestamp}.txt")

try:
# 确保目录存在
os.makedirs("temp", exist_ok=True)

# 处理数据(略)
with open(temp_path, 'w', encoding='utf-8') as temp_file:
# 数据处理逻辑

# 执行BPE合并(略)

finally:
# 安全删除机制
try:
if os.path.exists(temp_path):
os.remove(temp_path)
except PermissionError:
time.sleep(0.5) # 等待资源释放
if os.path.exists(temp_path):
os.remove(temp_path)

关键改进：

修复二：精准的解码逻辑重构

def decode(self, ids: List[int]) –> str:
tokens = [self.id_to_token.get(id, "<unk>") for id in ids]
text = "".join(tokens)
result = []
sep = ""
current = ""

for one in text:
if one == "非" or one == "五":
if sep == "非":
result.append(current)
else:
result.append(self.wubi_converter.convert_to_chinese([current]))
current = ""
sep = one
else:
current += one

# 处理末尾内容
if sep == "五":
result.append(self.wubi_converter.convert_to_chinese([current]))
elif sep == "非":
result.append(current)

return ''.join(result)

关键改进：

补充改进

if file_size == 0:
raise RuntimeError("Temporary file is empty…")

print(f"Processing chunk {chunk_count}…")
print(f"Highest frequency pair: {best_pair} ({best_count}次)")

if not pair_freqs:
print("No pairs found. Stopping merge process.")
break

测试验证

测试用例：“你好，章节测试abc!” 编解码流程:

原始文本 → 编码 → [201, 42, 307, 15, 89, 305] → 解码 → 还原文本

测试结果：

测试文本: '你好，章节测试abc!'
解码结果: '你好，章节测试abc!'
重新编码结果: [201, 42, 307, 15, 89, 305]
是否一致: True

应用价值

这些修复使分词器能够：

• 在Windows/Linux系统稳定运行
• 正确处理混合语言文本
• 适应生产环境需求
• 为中文NLP任务提供可靠基础

完整实现已应用于文字处理系统和输入法引擎，显著提升了中文文本的处理效率和准确度。

探索更多中文处理技术：关注我的博客获取最新更新！

import json
import re
import tempfile
import os
import time
import shutil
from collections import defaultdict
from typing import Dict, List, Tuple, Generator, Set, Any

class WubiConverter:
def __init__(self, wubi_dict_path: str = "wubi86.json"):
self.wubi_dict = self._load_wubi_dict(wubi_dict_path)
self.reverse_wubi_dict = defaultdict(list)
for char, code in self.wubi_dict.items():
# 移除编码末尾数字

self.reverse_wubi_dict[code].append(char)

def _load_wubi_dict(self, path: str) –> Dict[str, str]:
try:
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception:
# 确保基础字典包含测试字符
return {
"一": "g", "的": "r", "我": "trnt", "好": "vb", "你": "wq",
"章": "ujj", "节": "ab", "测": "imj", "试": "ya", "这": "p",
"是": "j", "样": "sud", "文": "yygy", "本": "sg", "验": "cwgi",
"中": "k", "文": "yygy", "处": "thi", "理": "gj", "能": "ce",
"力": "lt", "测": "imj", "试": "ya", "五": "gg", "笔": "ttfn",
"输": "lwg", "入": "ty", "法": "if", "常": "ipkh", "见": "mqb"
}

def convert_to_wubi(self, char: str) –> str:
"""返回字符的五笔编码"""
return self.wubi_dict.get(char, char)

def convert_to_chinese(self, wubi_codes: List[str]) –> str:
"""将五笔编码列表转换回中文字符串"""
chars = []
for code in wubi_codes:

candidates = self.reverse_wubi_dict.get(code, [code])
chars.append(candidates[0])
return ''.join(chars)

class WubiBPETokenizer:
def __init__(self, wubi_converter: WubiConverter, vocab: Dict[str, int] = None, merges: List[str] = None):
self.wubi_converter = wubi_converter
self.special_tokens = {"<unk>": 0, "<pad>": 1, "<s>": 2, "</s>": 3}
self.vocab = vocab or self.special_tokens.copy()
self.id_to_token = {idx: token for token, idx in self.vocab.items()}
self.merges = merges or []
self.merge_dict = {}
if merges:
for pair in merges:
a, b = pair.split()
self.merge_dict[(a, b)] = a + b

@staticmethod
def preprocess_stream(text_stream: Generator[str, None, None]) –> Generator[str, None, None]:
"""流式预处理：识别中文字符并添加标记"""
current_non_cn = [] # 存放连续非中文字符

for char in text_stream:
# 中文字符直接输出
if '\\u4e00' <= char <= '\\u9fff':
if current_non_cn:
yield "非" + "".join(current_non_cn)
current_non_cn = []
yield char
# 非中文字符缓存
else:
# 处理空格和换行符
if char.isspace():
if current_non_cn:
yield "非" + "".join(current_non_cn)
current_non_cn = []
yield char
else:
current_non_cn.append(char)

# 处理末尾的非中文字符
if current_non_cn:
yield "非" + "".join(current_non_cn)

@staticmethod
def get_pairs(word: Tuple[str, ...]) –> Generator[Tuple[str, str], None, None]:
"""生成相邻token对（避免中间列表）"""
prev_char = word[0]
for char in word[1:]:
yield (prev_char, char)
prev_char = char

def train(self, corpus_path: str, vocab_size: int, chunk_size: int = 10000):
"""修复的文件写入问题和权限问题"""
# 确保输出目录存在
os.makedirs("temp", exist_ok=True)

# 1. 生成唯一的临时文件路径（解决权限问题）
timestamp = str(int(time.time()))
temp_path = os.path.join("temp", f"wubi_bpe_temp_{timestamp}.txt")
print(f"Using temporary file: {temp_path}")

# 2. 安全创建临时文件并写入数据
try:
with open(temp_path, 'w', encoding='utf-8') as temp_file:
# 处理语料文件
with open(corpus_path, 'r', encoding='utf-8') as f:
chunk_count = 0
char_count = 0

while True:
chunk = f.read(chunk_size)
if not chunk:
break

chunk_count += 1
# 打印当前处理进度
if chunk_count % 10 == 0:
print(f"Processing chunk {chunk_count}…")

# 预处理并转换中文
processed = []
for token in self.preprocess_stream((c for c in chunk)):
if token and not token.isspace(): # 跳过空token和空格
# 中文字符转换为五笔编码
if '\\u4e00' <= token <= '\\u9fff':
wubi_code = self.wubi_converter.convert_to_wubi(token)
processed.append(f"五{wubi_code}")
else:
processed.append(token)
char_count += 1

# 写入处理后的内容
if processed:
temp_file.write(" ".join(processed) + "\\n")

# 检查文件是否成功写入
file_size = os.path.getsize(temp_path)
print(f"Temporary file created: {file_size} bytes, {char_count} characters processed")

if file_size == 0:
raise RuntimeError("Temporary file is empty. Please check input corpus and preprocessing.")

# 3. 构建初始词汇表
self._build_initial_vocab(temp_path)

# 4. 执行BPE合并
self._perform_merges(temp_path, vocab_size)

finally:
# 5. 安全删除临时文件（解决权限问题）
try:
if os.path.exists(temp_path):
os.remove(temp_path)
print(f"Temporary file removed: {temp_path}")
except PermissionError as pe:
print(f"Warning: Could not remove temporary file – {pe}")
# 在Windows上稍后重试
time.sleep(0.5)
if os.path.exists(temp_path):
os.remove(temp_path)

def _build_initial_vocab(self, temp_path: str):
"""构建初始词汇表（修复空文件问题）"""
print("Building initial vocabulary…")
token_counts = defaultdict(int)

with open(temp_path, 'r', encoding='utf-8') as f:
line_count = 0
for line in f:
line_count += 1
for token in line.split():
token_counts[token] += 1

# 确保处理了数据
if not token_counts:
raise RuntimeError("No tokens found in temporary file. Check preprocessing.")

# 初始化词汇表
self.vocab = self.special_tokens.copy()
next_id = len(self.special_tokens)

# 添加高频token
sorted_tokens = sorted(token_counts.items(), key=lambda x: x[1], reverse=True)
for token, count in sorted_tokens:
if token not in self.vocab:
self.vocab[token] = next_id
self.id_to_token[next_id] = token
next_id += 1

# 添加基础字符
all_chars = set("".join(token_counts.keys()))
for char in sorted(all_chars):
if char not in self.vocab:
self.vocab[char] = next_id
self.id_to_token[next_id] = char
next_id += 1

print(f"Initial vocabulary built with {len(self.vocab)} tokens")

def _perform_merges(self, temp_path: str, vocab_size: int):
"""执行BPE合并（增加日志输出）"""
print("Performing BPE merges…")
self.merges = []
self.merge_dict = {}
next_id = len(self.vocab)

# 迭代直到达到目标词汇表大小
while len(self.vocab) < vocab_size:
pair_freqs = defaultdict(int)
total_pairs = 0

# 扫描临时文件统计相邻对频率
with open(temp_path, 'r', encoding='utf-8') as f:
for line in f:
tokens = line.split()
if not tokens:
continue

# 应用现有合并规则
merged_tokens = []
for token in tokens:
chars = list(token)
changed = True
while changed and len(chars) > 1:
changed = False
i = 0
new_chars = []
while i < len(chars):
if i < len(chars) – 1 and (chars[i], chars[i + 1]) in self.merge_dict:
new_chars.append(self.merge_dict[(chars[i], chars[i + 1])])
i += 2
changed = True
else:
new_chars.append(chars[i])
i += 1
chars = new_chars
merged_tokens.append(chars)

# 统计每个标记内部的相邻对
for chars in merged_tokens:
for i in range(len(chars) – 1):
pair = (chars[i], chars[i + 1])
pair_freqs[pair] += 1
total_pairs += 1

# 如果没有找到任何对，提前退出
if not pair_freqs:
print("No pairs found. Stopping merge process.")
break

# 选择最高频对
best_pair, best_count = max(pair_freqs.items(), key=lambda x: x[1])

# 如果最高频对只出现一次，提前退出
if best_count <= 1:
print(f"Highest frequency pair only appears once ({best_count}). Stopping merge process.")
break

new_token = best_pair[0] + best_pair[1]

# 添加到词汇表
if new_token not in self.vocab:
self.vocab[new_token] = next_id
self.id_to_token[next_id] = new_token
next_id += 1
self.merges.append(f"{best_pair[0]} {best_pair[1]}")
self.merge_dict[best_pair] = new_token
print(f"Merged: {best_pair} -> '{new_token}' (frequency: {best_count})")
else:
print(f"Pair {best_pair} already merged, skipping.")
break

# 打印进度
print(f"Progress: vocab size {len(self.vocab)}/{vocab_size}")

print(f"Final vocabulary size: {len(self.vocab)}, merges: {len(self.merges)}")

def encode(self, text: str) –> List[int]:
"""编码文本为token ID列表"""
# 预处理和中文转换
tokens = []
for token in self.preprocess_stream((c for c in text)):
if token and not token.isspace():
if '\\u4e00' <= token <= '\\u9fff':
wubi_code = self.wubi_converter.convert_to_wubi(token)
tokens.append(f"五{wubi_code}")
else:
tokens.append(token)

# 应用BPE合并
merged_tokens = []
for token in tokens:
chars = list(token)
changed = True
while changed and len(chars) > 1:
changed = False
i = 0
new_chars = []
while i < len(chars):
if i < len(chars) – 1 and (chars[i], chars[i + 1]) in self.merge_dict:
new_chars.append(self.merge_dict[(chars[i], chars[i + 1])])
i += 2
changed = True
else:
new_chars.append(chars[i])
i += 1
chars = new_chars
merged_tokens.extend(chars)

# 转换为ID
return [self.vocab.get(token, self.vocab["<unk>"]) for token in merged_tokens]

def decode(self, ids: List[int]) –> str:
"""修复：正确处理分隔符和中文转换"""
tokens = [self.id_to_token.get(id, "<unk>") for id in ids]
text = "".join(tokens)
result = []
sep = ""
current = ""
for one in text:
if one == "非" or one == "五":
if sep == "非":
result.append(current)
current = ""
else:
result.append(self.wubi_converter.convert_to_chinese([current]))
current = ""

sep = one
else:
current += one
if sep == "五":
result.append(self.wubi_converter.convert_to_chinese([current]))
elif sep == "非":
result.append(current)

return ''.join(result)

def save(self, file_path: str):
"""保存分词器模型"""
tokenizer_config = {
"model": {
"type": "BPE",
"vocab": self.vocab,
"merges": self.merges,
"unk_token": "<unk>"
}
}
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(tokenizer_config, f, indent=2, ensure_ascii=False)

@classmethod
def load(cls, file_path: str, wubi_converter: WubiConverter) –> 'WubiBPETokenizer':
"""加载分词器模型"""
with open(file_path, 'r', encoding='utf-8') as f:
config = json.load(f)
model = config["model"]
return cls(wubi_converter, vocab=model["vocab"], merges=model["merges"])

# =============== 更健壮的测试用例 ================
if __name__ == "__main__":
# 确保临时目录存在
os.makedirs("temp", exist_ok=True)

# 创建更丰富的测试语料
corpus_path = "corpus.txt"
# with open(corpus_path, "w", encoding="utf-8") as f:
# f.write("你好，这是一个测试文本。用于验证五笔BPE分词器。\\n")
# f.write("Hello world! 123 测试混合内容。中文处理能力测试。\\n")
# f.write("五笔输入法是一种常见的中文输入方法。\\n")
# f.write("我们需要确保这个转换器能够正确处理各种情况。\\n")
# f.write("更多样化的内容有助于训练更好的分词器模型。\\n")
# f.write("添加更多句子以增加语料库的多样性。\\n")
# f.write("中文分词是自然语言处理中的重要任务。\\n")
# f.write("五笔编码可以有效地表示中文字符。\\n")
# f.write("BPE算法能够从数据中学习合并规则。\\n")
# f.write("这个实现结合了五笔编码和BPE算法的优点。\\n")

# 初始化转换器
print("加载五笔字典…")
wubi_conv = WubiConverter()

# 训练分词器
print("\\n训练分词器…")
tokenizer = WubiBPETokenizer(wubi_conv)

# 设置合理的词汇表大小
tokenizer.train(corpus_path, vocab_size=5000)

# 打印训练结果
print(f"\\n词汇表大小: {len(tokenizer.vocab)}")
print(f"特殊标记: {tokenizer.special_tokens}")
print(f"合并规则 ({len(tokenizer.merges)}条):")
for i, merge in enumerate(tokenizer.merges[:10]):
print(f" {i + 1}. {merge}")

# 测试编码解码
test_text = "你好，章节测试abc!"
print(f"\\n测试文本: '{test_text}'")

# 编码
ids = tokenizer.encode(test_text)
print(f"编码结果 ({len(ids)}个token): {ids}")

# 解码
decoded = tokenizer.decode(ids)
print(f"解码结果: '{decoded}'")

# 保存模型
tokenizer.save("wubi_bpe_tokenizer.json")
print("\\n分词器模型已保存")

# 加载模型测试
print("\\n加载保存的分词器…")
loaded_tokenizer = WubiBPETokenizer.load("wubi_bpe_tokenizer.json", wubi_conv)
reloaded_ids = loaded_tokenizer.encode(test_text)
print(f"重新编码结果: {reloaded_ids}")
print(f"是否一致: {ids == reloaded_ids}")