简介：面向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 自动机而不是 contains

2.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 root

setdefault在建树时如果字符不存在就创建新节点，这步把 Trie 的插入逻辑压成了一行。建 fail 指针时必须按层遍历，所以用了deque做广度优先搜索。child.output += child.fail.output的作用是把失败节点上命中的词也合并进来，比如匹配到「AB」时，如果「B」本身也是敏感词，它也能被收集到。这里的words就是词库列表，建议在传入前先去重、去空字符串，否则空词会让while循环出现死循环。

实际工程里，如果节点数量上万，Python 对象本身的内存开销会偏高，可以考虑用数组代替字典来存 children，但那是另一个优化故事了。先跑通这个版本，再根据量级决定要不要做压缩。

2.3 命中之后的处置策略：replace、mark 与 block 怎么选

匹配引擎只负责回答「命中了什么」，命中后怎么处置是策略层的事，我一般会把处置方式拆成三档。第一档是替换（replace），把敏感词替换成*或x，用于评论展示、昵称这类需要保留可读性的场景；第二档是标记（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.py

base_words.txt放公开词库转换出来的基础词，这份文件只在用脚本更新词库时被覆盖，平时不要手动改。custom_words.txt放业务自己的扩展词，这个文件是运营和开发日常维护的地方。分开存的原因很实际：公开词库下一次更新往往是一整份文件替换，如果手工改过 base 文件，合并时会非常痛苦，轻则丢词，重则把格式改坏导致加载失败。custom 文件保持独立，让更新 base 变成一个完全无脑的覆盖操作。

初始化加载的代码放到automaton.py里，复用第 2 章的build_automaton：

from 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(encoding="utf-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_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, maxsplit=1)[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(encoding="utf-8") except UnicodeDecodeError: text = fp.read_text(encoding="gbk") for line in text.splitlines(): word = clean_line(line) if word and word not in seen: seen.add(word) words.append(word) words.sort(key=len, reverse=True) out_file.write_text("\n".join(words), encoding="utf-8") print(f"total: {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、级别为 value，custom 覆盖 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(length=200): return "".join(random.choices(string.ascii_letters + string.digits, k=length)) def benchmark(f, text, rounds=10000): start = time.perf_counter() for _ in range(rounds): f.find_all(text) elapsed = time.perf_counter() - start print(f"rounds={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 飙到秒级，用户端就会明显卡顿。我现在的固定流程是：动词库必跑抽测，动匹配逻辑必跑压测，两条都过了才允许上线。这个习惯帮我挡掉过好几次「词库更新完接口超时」的线上事故，希望帮到你。

本文还有配套的精品资源，点击获取