短Token过滤
什么是短token(what)
RAG中的短token有两种含义
- 短文档块(chunk),在文档分块的时候因为文档太长,导致被分块嵌入而留下的尾巴
- 短Token,例如 “的”,“是”,这些无实际意义但却会产生语义近似的短词
深入理解中文分词原理
在讨论短Token识别方法之前,我们需要了解中文分词的基础原理。中文文本没有天然的分词边界,分词质量直接影响后续的短Token识别效果。
jieba分词原理
jieba是目前最流行的中文分词器,采用三层算法叠加策略:
第一层:前缀词典匹配(机械分词)
jieba构建基于Trie树的前缀词典,采用最大正向匹配算法(MM)进行分词。
class TrieNode: def __init__(self): self.children = {} self.is_end = False self.word = None第二层:HMM新词发现
对于未登录词,jieba引入隐马尔可夫模型(HMM)进行新词发现,采用B/M/E/S状态标注:
- B (Begin): 词开始
- M (Middle): 词中间
- E (End): 词结束
- S (Single): 单字词
第三层:Viterbi算法解码
动态规划求解最优状态序列:
δ_t(i) = max[P(状态i | t时刻) × max[δ_{t-1}(j) × A_{ji}]]主流中文分词器对比
| 分词器 | 算法特点 | 速度 | 准确率 | 适用场景 |
|---|---|---|---|---|
| jieba | 前缀词典+HMM | 快 | 较高 | 通用场景 |
| pkuseg | CNN+HMM | 中 | 高 | 高精度需求 |
| LTP | 神经网络 | 中 | 高 | 复杂NLP任务 |
| HanLP | 多种算法融合 | 中 | 高 | 企业级应用 |
| thulac | CRF+前缀词典 | 快 | 较高 | 平衡速度与精度 |
分词算法分类
机械分词法:正向最大匹配(MM)、反向最大匹配(RMM)、双向最大匹配
N-gram语言模型:通过统计词序列共现概率解决分词歧义
序列标注方法:HMM、MEMM、CRF使用BEMS标签体系进行分词
深度学习方法:BiLSTM-CRF、BERT-CRF提供最高精度
jieba实战使用
import jieba
# 精确模式words = jieba.lcut("我正在学习自然语言处理", cut_all=False)# ['我', '正在', '学习', '自然语言', '处理']
# 全模式words_all = jieba.lcut("我正在学习自然语言处理", cut_all=True)
# 关键词抽取import jieba.analysekeywords = jieba.analyse.extract_tags(text, topK=10)如何处理短token
● RAG中短Token的处理策略
一、短Token的识别
定义标准:
┌────────────┬────────────────────┬──────────────────┐ │ 类型 │ 特征 │ 示例 │ ├────────────┼────────────────────┼──────────────────┤ │ 停用词 │ 长度≤2,无语义信息 │ 的、是、在、和 │ ├────────────┼────────────────────┼──────────────────┤ │ 高频通用词 │ 文档频率>阈值 │ 我们、可以、这个 │ ├────────────┼────────────────────┼──────────────────┤ │ 模糊词 │ 单字、无义词 │ 啊、嗯、什么 │ ├────────────┼────────────────────┼──────────────────┤ │ 标点/数字 │ 纯符号或数字 │ 123、@#$ │ └────────────┴────────────────────┴──────────────────┘识别方法:
方法1:预设停用词表
STOP_WORDS = {‘的’, ‘是’, ‘在’, ‘和’, ‘了’, ‘我’, ‘你’, ‘它’, ‘这’, ‘那’}
方法2:IDF筛选(信息量视角)
# IDF = log(N / df) → IDF过低 = 区分度低 = 信息量少
def is_short_token(token, idf_threshold=2.0): """IDF低于阈值视为短token""" return get_idf(token) < idf_threshold方法3:混合策略
def is_short_token_v2(token, min_len=2, idf_threshold=2.0): return len(token) <= min_len or get_idf(token) < idf_threshold方法4:多维度组合过滤(综合实现)
def comprehensive_filter(text, min_chars=10, min_tokens=3, max_stopword_ratio=0.6, min_entropy=2.0, max_punct_ratio=0.3): if len(text) < min_chars or len(text.split()) < min_tokens: return False if not filter_by_stopword_ratio(text, max_stopword_ratio): return False if not filter_by_entropy(text, min_entropy): return False if not filter_by_punctuation_ratio(text, max_punct_ratio): return False return True
def filter_by_stopword_ratio(text, max_ratio=0.6): import jieba words = list(jieba.cut(text)) stopwords = {'的', '了', '是', '在', '有', '和', '就', '不', '人', '都'} stopword_count = sum(1 for w in words if w in stopwords) ratio = stopword_count / len(words) if words else 1.0 return ratio < max_ratio
def filter_by_entropy(text, min_entropy=2.0): import math from collections import Counter if not text: return False char_freq = Counter(text) length = len(text) entropy = 0.0 for count in char_freq.values(): p = count / length if p > 0: entropy -= p * math.log2(p) return entropy >= min_entropy
def filter_by_punctuation_ratio(text, max_punct_ratio=0.3): import re punct_count = len(re.findall(r'[,。!?、;:""''()【】《》\-—…]', text)) total_chars = len(text) if total_chars == 0: return False return punct_count / total_chars < max_punct_ratio二、过滤策略(Query层面)
1. 直接过滤
# 检索前从query中移除短token
def filter_query(query: str) -> str: tokens = tokenize(query) filtered = [t for t in tokens if not is_short_token(t)] return " ".join(filtered)
# 例: "Java是什么" → "Java"# 例: "如何在Python中使用多线程" → "Python使用多线程"
优点:简单直接,检索效率高缺点:可能丢失部分语义(如"不Java")2. 保留但降权
def weighted_query(tokens): """ 短token给予极低权重 长token(实体、术语)给予高权重 """ weights = [] for t in tokens: if is_short_token(t): weights.append(0.1) # 短token权重降为0.1 else: weights.append(1.0) # 正常token权重1.0 return weights
# 向量检索时:query_vector = Σ(token_vector * weight)三、Query改写策略
1. 同义词/概念扩展
# 短token "大" 可能需要扩展 "大" → "大公司" / "大型" / "规模大"
# 使用词向量找相似长词 def expand_short_token(token): similar = find_similar_tokens(token, topk=3) # 过滤掉仍然是短token的 return [t for t in similar if len(t) > 2]
2. 伪相关反馈 (Pseudo Relevance Feedback)
def query_expansion(query, top_k=5): """ 1. 用原query检索,获取Top-K结果 2. 从结果中提取高频实词 3. 将实词加入原query """ initial_results = vector_search(query, top_k=top_k) expanded_terms = extract_keywords_from_results(initial_results) # 只保留非短token的关键词 expanded_terms = [t for t in expanded_terms if not is_short_token(t)] return query + " " + " ".join(expanded_terms)四、短Token命中判定(结果质量评估)
这是用于评估检索结果可信度的机制,不改变检索本身。
命中检测流程
Step 1: 从Query中提取短token集合 S = {s1, s2, ...}
Step 2: 对每个检索结果R,计算短token命中率 hit_ratio(R) = |S ∩ content(R)| / |S|
Step 3: 置信度校准 if hit_ratio > 0.8: # 几乎所有短token都命中了 confidence *= 0.5 # 降低该结果可信度
Step 4: 综合排序 final_score(R) = similarity_score(R) * confidence(R) 代码示例
def calculate_short_token_hit_ratio(query: str, doc_content: str) -> float: """ 计算短token命中率 命中率过高说明文档"假相关" """ query_tokens = set(tokenize(query)) doc_tokens = set(tokenize(doc_content))
short_tokens = {t for t in query_tokens if is_short_token(t)} if not short_tokens: return 0.0
# 命中的短token hit_tokens = short_tokens & doc_tokens return len(hit_tokens) / len(short_tokens)
def adjust_confidence(sim_score: float, hit_ratio: float) -> float: """ 短token命中过高 → 降权 """ if hit_ratio > 0.8: return sim_score * 0.3 # 严重降权 elif hit_ratio > 0.5: return sim_score * 0.7 # 轻度降权 return sim_score判定场景示例
┌──────────────────┬─────────────┬────────────┬────────────┬───────────────┐ │ Query │ 短Token集合 │ Doc A 命中 │ Doc B 命中 │ 判定 │ ├──────────────────┼─────────────┼────────────┼────────────┼───────────────┤ │ "什么是AI" │ {什么,是} │ 100% │ 60% │ Doc B更可信 │ ├──────────────────┼─────────────┼────────────┼────────────┼───────────────┤ │ "Java多线程实现" │ {} │ - │ - │ 无短token干扰 │ └──────────────────┴─────────────┴────────────┴────────────┴───────────────┘五、实战方案总结
┌─────────────────────────────────────────────────────┐ │ 完整处理流程 │ ├─────────────────────────────────────────────────────┤ │ │ │ Query: "Java多线程是什么" │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 1. 短Token识别 │ → {什么,是} │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 2. 过滤/降权 │ → query = "Java多线程" │ │ │ │ weight = {Java:1, 多线程:1, 什么:0.1, 是:0.1} │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 3. 向量检索 │ → Top-10 结果 │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 4. 命中判定 │ → 计算每个结果的短token命中率 │ │ │ (置信度校准) │ → 命中过高则降权 │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ 最终结果 │ │ │ └─────────────────────────────────────────────────────┘关键原则
┌───────────┬─────────────────────────────────────────┐ │ 阶段 │ 原则 │ ├───────────┼─────────────────────────────────────────┤ │ 识别 │ 长度 + IDF双重判定,不依赖单一标准 │ ├───────────┼─────────────────────────────────────────┤ │ Query改写 │ 短token扩展为相关概念,而非直接删除语义 │ ├───────────┼─────────────────────────────────────────┤ │ 命中判定 │ 用于置信度校准,不是直接过滤结果 │ ├───────────┼─────────────────────────────────────────┤ │ 降权幅度 │ 渐进式(0.3/0.7/1.0),避免过度惩罚 │ └───────────┴─────────────────────────────────────────┘追加内容 (2026-04-07)
RAG系统中短Token过滤的最佳实践
在构建RAG(检索增强生成)系统时,短Token过滤是一个容易被忽视但至关重要的环节。本文将深入探讨短Token的识别方法、过滤策略以及实际应用中的注意事项。
什么是短Token
短Token是指语义信息量过低的文本块,主要表现为以下几类:
- 极短文本:如”是的”、“好的”、“OK”等无实质内容的回复
- 单独标点或数字:如”…”、“123”、”——“等
- 无意义短语:重复的模板文本、问候语等
- 空内容块:只有标签或占位符而没有实际语义
一般来说,低于3-5个Token或10-20个字符的chunk被认为是短Token。这一阈值并非固定值,需根据具体业务场景和分词器特性进行调整。
为什么需要过滤短Token
过滤短Token能为RAG系统带来多方面的提升:
降低噪声干扰:短Token容易产生虚假匹配,例如用户查询”的”可能会匹配到大量包含该字的文档,但这些匹配并无实际检索价值。
节省系统资源:减少无意义的向量存储和计算开销。在大规模向量数据库中,这一优化能显著降低成本。
提升检索质量:避免无意义的chunk占据检索Top-K位置,确保返回结果的相关性和实用性。
改善用户体验:返回更有实质内容的上下文,减少LLM处理无效信息的负担。
常用识别方法
长度阈值法
最直接的方法,根据字符数进行过滤:
def filter_by_length(text: str, min_chars: int = 10) -> bool: """判断文本是否低于长度阈值""" char_count = len(text.strip()) return char_count >= min_chars停用词比例法
检查停用词在文本中的占比,过高的停用词比例通常意味着内容空洞:
import jieba
STOPWORDS = {'的', '了', '是', '在', '我', '有', '和', '就', '不', '人'}
def stopword_ratio(text: str) -> float: """计算停用词比例""" words = list(jieba.cut(text)) if not words: return 1.0 stopword_count = sum(1 for w in words if w in STOPWORDS) return stopword_count / len(words)N-gram重复检测
通过检测重复模式识别模板化内容:
from collections import Counter
def has_repetitive_pattern(text: str, n: int = 3, threshold: float = 0.6) -> bool: """检测是否存在重复的N-gram模式""" words = text.split() if len(words) < n: return True ngrams = [tuple(words[i:i+n]) for i in range(len(words)-n+1)] counter = Counter(ngrams) if not counter: return True most_common_ratio = counter.most_common(1)[0][1] / len(ngrams) return most_common_ratio > threshold信息熵/复杂度法
利用信息熵检测文本复杂度,低熵值通常表示内容简单重复:
import mathfrom collections import Counter
def text_entropy(text: str) -> float: """计算文本的信息熵""" if not text: return 0.0 counter = Counter(text) length = len(text) entropy = 0.0 for count in counter.values(): p = count / length entropy -= p * math.log2(p) return entropy标点/数字比例法
检测标点符号和数字的占比,过高时可能为无效内容:
import re
def punctuation_ratio(text: str) -> float: """计算标点符号占比""" punctuation = set(',。!?、:;""''()【】《》—…·') if not text: return 0.0 punc_count = sum(1 for c in text if c in punctuation) return punc_count / len(text)IDF筛选法:信息量视角
IDF(逆文档频率)是衡量词区分度的重要指标:
IDF = log(N / df)
其中N为文档总数,df为包含该词的文档数。IDF值越低,表示该词的区分度越低,信息量越少。
高频停用词:如”的”、“了”、“是”等几乎出现在所有文档中,df接近N,IDF接近0,应过滤。
专业术语:出现在较少文档中,df较小,IDF较高,应保留。
小语料库下IDF的问题
然而,在小语料库场景下,IDF筛选存在显著局限:
问题示例:假设N=4个文档,“的”字只出现在1个文档中,则:
IDF(“的”) = log(4/1) = log(4) ≈ 1.386
这个值并不低,“的”反而不会被过滤掉。
根本原因:小语料库下,文档频率无法准确反映词的实际重要性。统计样本不足导致概率估计偏差。
解决方案:IDF筛选应与停用词表组合使用,形成双重保障:
def should_filter(token: str, idf_threshold: float = 0.5) -> bool: """综合判断是否应过滤""" # 命中停用词表 -> 直接过滤 if token in STOPWORDS: return True # IDF过低且非专业术语 -> 过滤 if get_idf(token) < idf_threshold and not is_domain_term(token): return True return False完整的过滤流程
实际应用中,短Token过滤应嵌入整个文档处理流水线:
文档 → 分块 → 分词 → 短Token识别 → 过滤规则 → 向量化存储具体实现时需注意以下几点:
-
阈值需调优:根据业务场景和分词器特性,选择合适的字符数/Token数阈值。
-
多策略组合:长度阈值、停用词比例、N-gram重复检测等方法组合使用,互相补充。
-
评估指标:过滤后检索 precision/recall 的变化是检验过滤策略有效性的最终标准。
-
可配置性:提供配置选项,让用户根据实际需求调整过滤强度。
总结
短Token过滤是RAG系统预处理阶段的重要环节,直接影响检索质量和系统效率。通过合理组合长度阈值法、停用词比例法、IDF筛选等多种策略,可以有效去除噪声内容,提升检索结果的相关性。在实际应用中,阈值选择和策略组合需要根据具体业务场景进行调优和验证。