敏感词过滤工程实践：AC自动机原理、词库处理与避坑指南

简介面向Go开发者的敏感词过滤组件基于DFA确定性有限自动机算法实现可用于聊天、评论、内容审核等场景的敏感词拦截与替换。词库暂用百度搜索到的通用列表可整体替换为其他词库也支持按需自行添加词条方便快速搭建符合业务要求的过滤能力。代码由Java版DFA算法实现直接转移而来结构简明。资源压缩包共6个文件包含4个Go源文件、1个词典文件和1个说明文档整体仅3KB阅读与维护都很轻量。Go源文件分别承担词典加载、敏感词匹配和替换等核心逻辑md文档介绍了加载与调用方式swf_test.go则给出可直接运行的示例。使用上在项目初始化时调用ReadSwfDict加载词库在检查点通过Match验证或用Repl替换敏感词逻辑清晰、易于二次开发。目前已有1907人学习下载适合希望掌握DFA过滤实现、或需要轻量级敏感词方案并自行扩展词库的Go开发者参考。1. Sensitive-word-filtering 这类敏感词过滤项目入手先想清楚「词库从哪来、扩展怎么加」Sensitive-word-filtering 这类敏感词过滤项目最常见的起点是找一份现成词库很多人的第一反应就是把网上流传的那份公开敏感词库直接下载下来加载进工程就当成完事了。词库确实能省掉从零整理基础词的力气但它不是银弹公开词库覆盖的是通用场景评论区里的黑话、行业特有的违禁信息、带变体的表达它基本管不到所以「可以自行扩展」才是这类项目真正进入生产环境的前提。这篇文章不堆理论直接拆三件事敏感词匹配引擎是怎么工作的、怎么把公开词库变成工程里稳定加载的词典、以及扩展词库时那些让人翻车的坑。适合正在做评论审核、聊天室过滤、表单内容校验的工程师也适合想低成本验证过滤方案是否可行的产品同学。下面从匹配原理说起这部分决定了你后面是每天被性能问题追着跑还是一劳永逸。2. 匹配引擎原理敏感词过滤为什么首选 AC 自动机而不是 contains2.1 从 contains 到字典树一次文本扫描的背后很多第一次写敏感词过滤的人代码长这样words [违禁词A, 违禁词B, 违禁词C] def filter_by_contains(text): hits [] for word in words: if word in text: hits.append(word) return hits这段逻辑本身没问题问题出在复杂度上。假设词库有 5000 个词输入文本平均 200 个字符最坏情况下要做 5000 次子串匹配也就是百万级字符比较。单次请求还能忍一旦接到线上接口QPS 稍微一起来CPU 直接被打满这就是最常见的第一个性能翻车点。要优化核心思路是改变「拿词去比文本」的方向改成「拿文本去查词」。把所有敏感词组织成一棵字典树Trie从文本的第一个字符开始往下走能走到某个词的结尾就说明命中。这样无论词库有多少词一次扫描能同时匹配出所有敏感词时间复杂度从 O(词数 × 文本长度) 降到接近 O(文本长度)。敏感词过滤领域里常说的 AC 自动机和 DFA 状态机本质上都是在这个思路上做的优化只是各自处理「失配后怎么跳」的方式不同。2.2 AC 自动机的完整构建过程先建 Trie再补 fail 指针AC 自动机在字典树的基础上加了一层失败指针fail pointer作用是在某个分支匹配失败时不用回到根节点重新开始而是跳到另一个可能的前缀分支继续匹配。这是它能保证线性扫描的关键下面的代码演示了完整的构建过程from collections import deque class ACNode: def __init__(self): self.children {} self.fail None self.output [] def build_automaton(words): root ACNode() # 第一步把每个词插入字典树 for word in words: node root for ch in word: node node.children.setdefault(ch, ACNode()) node.output.append(word) # 第二步用 BFS 补上每个节点的 fail 指针 queue deque() for node in root.children.values(): node.fail root queue.append(node) while queue: cur queue.popleft() for ch, child in cur.children.items(): queue.append(child) f cur.fail while f is not None and ch not in f.children: f f.fail child.fail f.children[ch] if f and ch in f.children else root if child.fail is not None: child.output child.output child.fail.output return rootsetdefault在建树时如果字符不存在就创建新节点这步把 Trie 的插入逻辑压成了一行。建 fail 指针时必须按层遍历所以用了deque做广度优先搜索。child.output child.fail.output的作用是把失败节点上命中的词也合并进来比如匹配到「AB」时如果「B」本身也是敏感词它也能被收集到。这里的words就是词库列表建议在传入前先去重、去空字符串否则空词会让while循环出现死循环。实际工程里如果节点数量上万Python 对象本身的内存开销会偏高可以考虑用数组代替字典来存 children但那是另一个优化故事了。先跑通这个版本再根据量级决定要不要做压缩。2.3 命中之后的处置策略replace、mark 与 block 怎么选匹配引擎只负责回答「命中了什么」命中后怎么处置是策略层的事我一般会把处置方式拆成三档。第一档是替换replace把敏感词替换成*或x用于评论展示、昵称这类需要保留可读性的场景第二档是标记mark命中后不打断用户操作只把内容和命中的词记录下来交给人工复核第三档是拦截block直接拒绝提交用于注册资料、支付备注这类一旦出现违规就必须挡住的场景。策略典型场景注意点replace评论区展示、弹幕注意替换后文本长度不变前端展示才不会错位mark私信、动态必须有异步审核流否则标记了没人看等于白做block注册、实名、支付拦截要带提示文案文案里不要回显敏感词这三个策略不是互斥的同一个词在不同场景可以有不同的处置级别。比如「广告词」在注册场景直接拦截在私信场景只标记。所以词库设计时最好把「词本身」和「处置级别」分开存储后面第 5 章会专门讲这个。理解了这一层再去看网上的过滤框架你就能一眼看出它到底只是封装了一个匹配库还是真正做了内容安全的完整方案。3. 本地跑通最小工程词库加载、命中查询、替换与三个关键参数3.1 最小工程目录基础词库和自定义词库为什么要分开我建议的目录结构是这样无论你最后用 Java、Go 还是 Python这个分层都通用sensitive-filter/ ├── dict/ │ ├── base_words.txt │ └── custom_words.txt ├── filter/ │ ├── __init__.py │ ├── automaton.py │ └── scanner.py ├── tests/ │ └── test_filter.py └── main.pybase_words.txt放公开词库转换出来的基础词这份文件只在用脚本更新词库时被覆盖平时不要手动改。custom_words.txt放业务自己的扩展词这个文件是运营和开发日常维护的地方。分开存的原因很实际公开词库下一次更新往往是一整份文件替换如果手工改过 base 文件合并时会非常痛苦轻则丢词重则把格式改坏导致加载失败。custom 文件保持独立让更新 base 变成一个完全无脑的覆盖操作。初始化加载的代码放到automaton.py里复用第 2 章的build_automatonfrom pathlib import Path from automaton import build_automaton class SensitiveFilter: def __init__(self, base_path: str, custom_path: str): words self._load_dict(base_path) self._load_dict(custom_path) self._root build_automaton(words) def _load_dict(self, path: str) - list[str]: words [] for raw in Path(path).read_text(encodingutf-8).splitlines(): word raw.strip() if word and not word.startswith(#): words.append(word) return words_load_dict做了两件事去掉行首尾的空格和换行跳过以#开头的注释行。这两个处理看起来不起眼实际能帮你避开大量「词没加载进来」的排查具体原因后面避坑章会讲。base_path和custom_path是词库文件的路径注意编码必须统一用 UTF-8如果是 Windows 下用记事本另存过的文件很可能变成带 BOM 的 UTF-8首字符会混进第一个词里让这个词永远匹配不上。3.2 命中查询与替换find_all 和 replace 的行为差异匹配接口我一般只暴露三个判断是否命中is_sensitive、取回所有命中详情find_all、替换replace。is_sensitive是拦截场景用的find_all是给人工审核系统用的replace是展示场景用的。三个接口共享同一套扫描逻辑只是返回值不同class SensitiveFilter: def find_all(self, text: str) - list[tuple[int, int, str]]: node self._root results [] for i, ch in enumerate(text): while node is not None and ch not in node.children: node node.fail node node.children[ch] if node and ch in node.children else self._root if node is None: node self._root continue for word in node.output: results.append((i - len(word) 1, i, word)) return results def is_sensitive(self, text: str) - bool: return bool(self.find_all(text)) def replace(self, text: str, repl: str *) - str: chars list(text) for start, end, _ in self.find_all(text): for pos in range(start, end 1): chars[pos] repl return .join(chars)find_all里每次失配先走node node.fail这个 while 就是 AC 自动机跳转的核心跳完之后如果还没有对应字符就回到根节点重新开始。命中结果里存的是(起始位置, 结束位置, 词)审核系统拿到这个三元组可以直接把命中片段高亮展示。replace遍历的是find_all的结果替换时把区间内每个字符都覆盖成repl所以即使两个词有重叠区间也不会出现漏替换。这里有个细节find_all的时间复杂度是 O(文本长度 命中次数)但如果一个文本里命中了大量敏感词results 列表会膨胀流量大的场景建议改成生成器或限制返回条数。3.3 三个必调参数ignore_case、max_word_len、replace_char不同实现里参数名可能不一样但这三个是决定过滤效果的分水岭。ignore_case控制是否忽略大小写默认关开启后会在加载词库和扫描文本时统一lower()max_word_len是单次扫描允许的最大词长超过这个长度的词条根本不建入 Trie用来挡词库里的超长垃圾数据replace_char是替换展示用的字符我一般用*也有团队用x或空格。参数建议值说明ignore_casefalse语言模型里英文变体多按需开启但要保证词库同步统一小写max_word_len32正常敏感词极少超过 32 字超长词大概率是脏数据replace_char*不要用空字符串替换后文本长度变化会导致前端展示错位ignore_case是我们当时被坑得最惨的一个参数。第一版没开用户把「ABCD」换成「abcd」就绕过了过滤后来改了匹配逻辑又因为词库里有大写字母导致原本能命中的词失效。所以我的习惯是词库在写入前统一转小写扫描时也统一转小写两端约定同一个规则比在匹配时做兼容要省心得多。4. 统一词库处理把公开词库和自定义词合并成一套稳定加载的词典4.1 公开词库的原始形态三个让你加载失败的问题从网上下载的公开敏感词库我见过的原始形态至少有五种有的一行一个词有的一行多个词用空格或逗号分隔有的带行号和注释有的是 GBK 编码。直接拿来加载第一个问题是编码报错Python 读 GBK 文件按 UTF-8 解必然抛异常第二个问题是行内会有多余内容比如「违禁词 注释」这种把注释一起算进词条等于引入了一批永远匹配不上的脏词第三个问题是重复词条极多一份几万行的词库去重后可能只剩几千条不去重的话 Trie 建出来会很大扫描性能白白受损。还有个容易被忽略的问题公开词库里混着大量「语义中性但在特定语境才有冒犯性」的词。这种词如果一股脑全进过滤词库误伤率会高到业务方直接找你所以转换时就要给词打上分级标记而不是只有「是敏感 / 不是敏感」两个状态。建议把词库分成三个级别直接拦截、仅标记、仅监控不展示。这样即使误伤也可以在标记阶段被人工放行而不是把用户内容直接吞掉。4.2 转换脚本清洗、去重、统一编码一次搞定下面这个脚本负责把散乱的原始词库文件转换成一份干净的、按长度倒序排列的标准词表from pathlib import Path import re def clean_line(line: str) - str | None: line line.strip() if not line or line.startswith(#) or line.startswith(//): return None line re.split(r[\s,、;], line, maxsplit1)[0] return line.strip() def convert(raw_dir: Path, out_file: Path) - None: seen set() words [] for fp in sorted(raw_dir.glob(*.txt)): try: text fp.read_text(encodingutf-8) except UnicodeDecodeError: text fp.read_text(encodinggbk) for line in text.splitlines(): word clean_line(line) if word and word not in seen: seen.add(word) words.append(word) words.sort(keylen, reverseTrue) out_file.write_text(\n.join(words), encodingutf-8) print(ftotal: {len(words)}) convert(Path(./raw_dict), Path(./dict/base_words.txt))脚本做了四件事编码探测、行清洗、去重、排序。try utf-8 except UnicodeDecodeError是处理混编码目录最简单可靠的办法遇到 GBK 文件自动降级。re.split只取第一段能同时处理空格、中文逗号、顿号、分号分隔的一行多词。按长度倒序排序是为了让长词排在前面方便你后续人工抽查高优先级词条以及在某些按顺序匹配的实现里保证长词优先命中。这里的raw_dir是原始词库文件所在目录out_file是输出路径注意输出文件的编码强制 UTF-8避免下次加载时又踩编码坑。4.3 扩展词的格式约定级别、场景、有效期custom 词库不建议再用纯文本一行一个词理由很简单你会需要知道这词是谁加的、什么级别、什么时候该清理。我习惯用 tab 分隔的四列格式违规词 1 comment 2025-06-01 某个品牌词 2 private_msg 2025-06-01四列分别是词条、级别1 直接拦截 / 2 仅标记、适用场景、生效日期。加载时按场景过滤比如私信场景只加载 level 1 和 2 中 scene 包含private_msg的词评论场景只加载 scene 为comment的词。这个约定的价值在运维阶段才体现得出来运营每周加词几个月后词库膨胀如果没有级别和场景维度你根本不敢删任何词因为不知道删了会影响哪个业务。加载合并时的规则是base 词库在前custom 词库在后如果同一个词在两边都出现以 custom 的级别为准。这样设计是为了让运营可以「降级」一个误伤的通用词而不需要去改 base 文件。合并逻辑在_load_dict里需要改成带去重的形式用一个 dict 以词为 key、级别为 valuecustom 覆盖 base 的同名词条。5. 避坑 / 排查敏感词过滤最容易翻车的五个时刻5.1 现象词库里明明有这个敏感词但文本就是匹配不到词存在、代码逻辑没问题、测试用例也过了但线上就是放过去了。这种「看了半天看不出原因」的问题八成是字形伪装用户用全角字符、大小写混写、中间插入零宽字符来绕过匹配。比如把「abc」写成「」全角形式ASCII 的匹配逻辑根本认不出。原因就是扫描前没有做输入归一化。解决方法是匹配前统一做一次文本预处理全角转半角、大写转小写、去掉零宽字符和不可见控制字符。归一化函数建议在过滤器外层调用不要把逻辑散落在各个业务代码里。当时我们因为这个漏放过一轮风险内容后来把归一化写进了过滤器的统一入口才从根上堵住。5.2 现象误伤率太高业务方开始投诉公开词库的通用词一旦加载误伤几乎是必然的。常见的例子是一些语气词、缩写、名词在部分语境下有冒犯含义但在正常对话里只是个普通词。原因就是词库没有分级把「任何时候都该拦」和「特定语境才该拦」的词混在一起了。解决思路是引入三级处置硬拦截只留给绝对敏感的少数词其余全部走标记和人工抽审。同时给 custom 词库加白名单机制如果一个词被反复误伤运营可以在白名单里维护例外规则比如「词 前后文特征」的组合才触发。只靠词库本身想 100% 不误伤是不可能的必须有人工闭环和可调节的白名单机制。5.3 现象加载几万词之后服务启动变慢到不可接受Trie 本身构建很快慢的往往是对原始文本逐行做字符串处理的过程。我见过一个案例词库文件 8 万行启动时要逐行 strip、编码转换、去重、排序加建树一趟跑下来要十几秒发布时每次都卡在启动检查。原因就是每次启动都在重复做清洗和建树而这个结果根本没有变的必要。解决办法是把构建好的自动机序列化落盘比如 Python 用 pickle 或 MessagePack 缓存Java 可以直接把对象序列化成文件词库文件不变就加载缓存变了才重新构建。缓存后启动时间可以从十秒级降到百毫秒内代价是发布时要记得带上词库版本号避免缓存和词库不一致。5.4 现象新加的敏感词怎么都不生效看起来是加词了、也提交了但线上就是不拦。排查时先看加载日志里这个词有没有被读进去很多实现会因为 BOM、空行、注释格式把一个词静默丢弃。再看是不是命中率统计的问题也就是词加载了但扫描时被max_word_len之类参数过滤掉了超长词在建树阶段就被丢掉日志又不报错看起来就像「加了没反应」。最重要的排查点是缓存如果服务是多实例部署每个实例各自持有一份内存词库你只更新了其中一个实例的词库文件其他实例自然还是旧数据。解决方法是给词库加一个版本号或更新时间戳启动时比对不一致才重新加载这样能避免线上不同实例行为不一致的问题。5.5 现象开放词库本身太脏统计报表没法看另一个常见问题是词库里大量内容是重复的、过时的甚至带乱码导致你统计命中率、TOP 敏感词时看到的最前面几条全是垃圾数据。原因就是很多公开词库是不同来源粗拼接出来的没有统一清洗。我处理这类词库时会在转换脚本里加一条规则长度小于 2 的词、包含不可见字符的词、只有标点符号的词一律丢弃。这些词对准确率没有贡献只会污染统计数据。每次从上游同步词库都要执行同一套清洗再去重不要等运行时才做这些事。6. 上线前的验证习惯让过滤器的准确率和性能都可度量6.1 先做人工抽测算准「误伤率」而不是只看「拦截率」验证过滤器不能只跑几个正向用例就完事线上真正的问题几乎都来自你没预料到的非敏感文本。我每次调整词库会先做一次抽测从线上历史数据里随机抽 1000 条内容人工标注其中哪些是真正该拦的再用过滤器跑一遍对比标注和命中的差异。输出一张结果表里面至少包含四个字段内容ID、是否命中、标注结果、命中的词。然后按混淆矩阵算两个数没拦住的漏网率和被拦错的无辜率。只看拦截率是典型的自欺欺人因为把词库调得越激进拦截率越高但误伤率也在同步升高后者才是真正影响用户体验的指标。6.2 压测脚本看吞吐量也要看 P99 而不是平均值性能验证我给一个可以直接跑的参考脚本它模拟固定长度文本上的批量匹配import random import string import time def gen_text(length200): return .join(random.choices(string.ascii_letters string.digits, klength)) def benchmark(f, text, rounds10000): start time.perf_counter() for _ in range(rounds): f.find_all(text) elapsed time.perf_counter() - start print(frounds{rounds}, elapsed{elapsed:.2f}s, qps{rounds / elapsed:.0f}) f SensitiveFilter(dict/base_words.txt, dict/custom_words.txt) benchmark(f, gen_text())压测最有价值的是确认两件事一是 QPS 是否能扛住你线上峰值二是替换和拦截接口不要因为命中大量词而退化到线性以上。平均值好看没有用要看 P99也就是最慢的 1% 请求花了多久内容审核这类接口一旦 P99 飙到秒级用户端就会明显卡顿。我现在的固定流程是动词库必跑抽测动匹配逻辑必跑压测两条都过了才允许上线。这个习惯帮我挡掉过好几次「词库更新完接口超时」的线上事故希望帮到你。本文还有配套的精品资源点击获取