简介：这是一份面向 Go 开发者的敏感词过滤工具源码包，基于 Java 版 DFA 算法迁移实现，内置百度分词词库，并支持自行扩展词条或替换为其他词库。资源共 6 个文件，包含 4 个 Go 源文件、1 个词典文件和 1 个 Markdown 说明文档，压缩包整体仅 3KB，结构轻量紧凑。核心功能覆盖词库加载、敏感词匹配与替换，可在项目 init 阶段调用 ReadSwfDict 载入词典，并在业务检查点通过 Match 或 Repl 方法完成校验与处理，swf_test.go 中还提供了明确的使用示例。对于需要快速接入内容审核、评论过滤或聊天消息检测的 Go 项目开发者，这份代码可以直接参考或复用，也可基于 DFA 算法进一步优化性能。目前已有 1907 人学习下载，适合中高级 Go 开发者阅读源码、扩展词库或改造成完整过滤服务。

1. 敏感词过滤：为什么“某度敏感词库+可扩展”才是落地的最短路径

做 UGC 内容审核时，敏感词过滤是最省资源的第一道闸。我刚接手某跨平台系统的消息审核时，天真地以为拉一份某度敏感词库，再用 Python 的in做字符串替换就能交差。结果上线第一天就翻车：用户正常说了一句“我要买某某”，被系统替换成满屏星号，投诉群直接炸了；同一天，一条绕过方案的真实违规内容却穿透了过滤器。后来才理解，敏感词过滤的真正难点不是“匹配”这个动作，而是词库的覆盖度、匹配算法的效率，以及业务要求的自行扩展能力。这套方案做的就是用某度敏感词库当基础底料，用 AC 自动机托住性能，再用自定义词库和规则让过滤器长出业务需要的牙齿。适合刚接内容安全、又不想第一步就接昂贵云审核服务的团队。

2. 某度敏感词库的获取与预处理：基础底料先洗干净

2.1 词库从哪来：如何拿到并评估某度敏感词库

某度敏感词库常见形态是每行一个词的文本文件，或者带分类的 JSON。早年在某代码托管平台能搜到整理好的镜像项目，格式一般如下：先是若干条来源说明，后面跟大批中文词、英文词、数字组合，夹杂着空行和重复项。我的选择标准很简单：优先找更新时间较近、词条数尽量多但明显经过人工整理的版本；宁可多源合并，也不只抱着单份文件当宝。

拿到词库后，第一步不是建索引，而是先做结构体检。用下面脚本统计一下词条总量、英文词占比、中文词占比、重复数量，以及词长分布。这些数字直接决定后续用哪种匹配算法、要不要分桶。

from pathlib import Path from collections import Counter import re path = Path("sensitive_words.txt") words = [line.strip().strip("\ufeff") for line in path.open(encoding="utf-8")] words = [w for w in words if w and not w.startswith("#")] total = len(words) unique = len(set(words)) has_chinese = sum(1 for w in words if re.search(r"[\u4e00-\u9fff]", w)) has_english = sum(1 for w in words if re.search(r"[A-Za-z]", w)) length_counter = Counter(len(w) for w in words) print("总词条:", total) print("去重后:", unique) print("含中文:", has_chinese, "含英文:", has_english) print("重复率: {:.2f}%".format((1 - unique / total) * 100 if total else 0)) print("词长分布 Top10:", length_counter.most_common(10))

逻辑说明：先按行读取并去掉 BOM，然后过滤注释和空行；分别统计中文和英文出现次数，注意一个词可能中英混合，所以两项相加可以大于总数。参数说明：path指向词库文件，如果不指定utf-8编码，Windows 下容易按 GBK 读乱码；统计词长分布可以帮你发现异常长词——如果出现几百字符的长串，多半是 URL 或带空格的整句，这类词在后续匹配时容易影响性能，需要单独处理。

选择词库时，我一般会把多个来源的词合并成候选集，再用 2.2 的清洗脚本统一。某个镜像项目可能几年没更新，直接使用的话，最新热词会漏得厉害，所以某度词库只能算底料，不能算全部。评估时还要看一眼词库的许可证和来源说明，有些镜像保留着原始抓取时间，这种可以作为时效性参考。

2.2 清洗、去重、编码统一：让某度敏感词库变成干净底料

这一节是最枯燥也最值得做的一步。原始词库常见的脏数据有：空行、BOM 头、全角空格、统一大小写后的重复项（如Fuck和fuck）、带 URL 的噪声、用中文逗号分隔导致一行被当成一个词。清洗目标只有一个：让每一行成为一个可被精确匹配的词条。

import re from pathlib import Path RAW_FILE = Path("raw_sensitive_words.txt") CLEAN_FILE = Path("sensitive_words_clean.txt") def normalize_word(word: str): word = word.strip().strip("\ufeff").strip() if not word or word.startswith("#"): return None # 统一英文大小写，中文不变 word = word.lower() # 全角空格转半角空格，避免匹配时看不见的坑 word = word.replace("\u3000", " ") # 连续多个空格压成一个 word = re.sub(r"\s+", " ", word) return word seen = set() with CLEAN_FILE.open("w", encoding="utf-8") as out_f: for line in RAW_FILE.open("r", encoding="utf-8", errors="ignore"): w = normalize_word(line) if not w or w in seen: continue # 跳过长度超过 50 的疑似 URL 或整段话 if len(w) > 50: continue seen.add(w) out_f.write(w + "\n") print("清洗完成，保留词条数:", len(seen))

逻辑说明：errors="ignore"会让无法按 utf-8 解码的字符被静默丢弃，这在处理半坏文件时比较稳妥，但也会丢掉一些生僻字；如果你确定词库原本是 GBK，更可控的做法是先尝试 utf-8，失败后用 gbk 解码再输出成 utf-8。参数说明：跳过超过 50 字符的词条是我自己的经验值，因为 URL 式噪声既难匹配又会拖慢构建；如果你的业务需要匹配长句，可以根据词长分布调整这个阈值。清洗脚本里用set去重，最终写回 utf-8 纯文本，方便后续程序和人工继续维护。

清洗之后再跑一次 2.1 的统计脚本，你会发现词条数可能缩水三分之一，但覆盖率反而更稳。因为重复和大小写变体占了大量冗余，不清洗直接建自动机，不仅匹配时会命中同一个词多次，扩展自定义词库时也不好判断到底加没加过。清洗后的文件建议纳入版本管理，每次更新词库时先用户 diff 确认变更量，避免某次脚本误删大量词条还不自知。

2.3 覆盖率与误杀率：用数据说话

只有词库没有验证指标，等于在裸奔。我会准备两组样本：一组是正常的业务语料（比如从历史日志里脱敏抽取的聊天记录、评论、昵称），一组是人为构造的违规样本。前者用来测误杀率，误杀率等于被错判为敏感词的正常消息数除以样本总数；后者用来测漏杀率，构造方式包括原始词条、大小写变体、符号插入、谐音变体等。

构造违规样本时，建议在每个原始词条上做几组变体：原词、全大写/全小写、中间插入一个空格、中间插入一个短横线、简体繁体互换。这部分样本数量不用多，每类几十条即可，但必须覆盖“符号插入”这类常见绕过手段。把样本保存成两个明文文件，后续每次改词库或改算法后都要重跑一遍，形成回归基线。这个习惯能救你很多次——我至少三次因为贪图省事没回归，结果上线后误杀率直接翻倍，那感觉比挖坑还难受。

3. 核心过滤算法选型：暴力匹配不行，AC 自动机才是真黑匣子

3.1 暴力遍历为什么扛不住上千条敏感词

很多人接手敏感词过滤的第一直觉是：for 遍历词库，对文本做in判断。如果词库只有几十条，这没问题；但某度词库经过清洗后通常也有几百到上千条。假设每条平均 4 个字符，每条消息 100 字符，一次请求要执行上千次子串搜索，Python 层面做的次数一多，单机 QPS 很容易被拖到两位数。更麻烦的是，你还要记录命中位置做替换，多个词命中同一条消息时要按偏移量排序，暴力遍历虽然也能做，但代码会越写越绕。

常见的生产方案有两类：基于 DFA 的状态机，和基于 AC 自动机的多模式匹配。DFA 思路是把所有词连成一棵前缀树，每个节点做字符跳转，匹配一遍文本就能找出所有词；AC 自动机则在 Trie 树上增加了 fail 指针，让匹配失败时能快速回退到最长后缀处继续走，相当于把多模式串匹配压成了线性扫描。以成熟度来看，我更推荐直接使用 AC 自动机，因为大部分实现已经在处理 fail 指针和编码问题，我们只需要关心词库和参数。如果非要用 DFA 做最小化状态机，内存确实更小，但实现复杂度高，自查成本也高，适合对内存极敏感的场景。

复杂度上，AC 自动机的构建时间是所有词总长度之和，匹配时间是纯线性的O(n)，与词条数量基本无关。这是它能支撑在线过滤的根本原因。有一个反直觉的地方：词库越大，构建时间越长，但匹配时间不会显著变慢。所以当你未来把自定义词库扩展到一万条，也不用担心单条请求的耗时，唯一的瓶颈是构建和内存。

3.2 用 pyahocorasick 构建敏感词匹配器

常见的 Python AC 自动机实现是pyahocorasick这个开源扩展，安装后直接构建和查询都很快。我一般封装成独立的匹配器模块，避免业务代码直接碰自动机对象。

import ahocorasick def build_automaton(word_list): automaton = ahocorasick.Automaton() for idx, word in enumerate(word_list): automaton.add_word(word, (idx, word)) automaton.make_automaton() return automaton def find_sensitive_words(text, automaton): hits = [] for end_index, (idx, word) in automaton.iter(text): start = end_index - len(word) + 1 hits.append((start, end_index, word)) return hits

逻辑说明：automaton.add_word的第二个参数是随命中一起返回的数据，我习惯存成(索引, 词)，这样后续做统计时能拿到是哪个词命中。make_automaton()完成后才能执行匹配，如果漏掉这一步，iter()会直接抛错。iter(text)返回的是匹配到的结束位置和 value，注意结束位置是闭区间，所以起点要减去词长再加 1。参数说明：文本越长，命中越多，返回的 hits 可能很长；如果只需要判断是否命中，可以用automaton.exists(text)替代，它遇到第一个命中就返回，能省不少时间。

实际使用时，不建议直接在业务逻辑里调用这个函数，而是把它放到一个类里，统一维护自动机的生命周期。代码里保存的idx也方便你在扩展词库时知道这个命中来自基础词库还是自定义词库，后续写统计报表时会很省事。另外特别注意：不要手动把中文encode成bytes再喂给自动机，默认字符串模式下中文按字符切分；一旦做字节级转换，中文会被切成三段的字节，匹配结果会完全失真。

3.3 匹配参数设计：大小写、空白、命中策略与内存

AC 自动机本身没有“大小写不敏感”的能力，所以要在喂文本前做归一化。我习惯在过滤服务里增加三个可配置参数：case_sensitive、skip_chars、max_hit_count。case_sensitive=False时，把输入文本和词库统一转为小写再匹配；skip_chars设置一个字符集合，用于在匹配前把常见的符号和空格替换成空字符，用来对抗“插字绕过”；max_hit_count控制最多返回多少个命中，防止恶意长文本塞进来后 hits 列表膨胀。

设计skip_chars时最容易忽略的问题是偏移量。如果先把文本中的符号删掉再做匹配，得到的命中位置是“归一化后文本”的位置，不能直接用来替换原始文本。我的做法是：在归一化阶段同时记录每个字符在原始文本中的真实索引，匹配完成后通过这个映射回填。这一步看着繁琐，但你在第 5 章会看到不做的后果。

命中策略上，AC 自动机默认返回所有匹配，同一个词可能在不同位置命中多次。如果业务只需要知道“有没有”，直接用exists()；如果需要展示给用户“命中了哪些词”，需要做一次按起点排序和区间合并，否则替换时很容易串位。一个隐藏的坑是：如果两个敏感词重叠，例如词库里有“保险”和“险种”，文本“保险种”会命中三个位置，这时候要决定按最左最长合并还是全部替换，我一般选择全部替换为相同数量的星号，优先保证不露敏感字。

4. 搭建敏感词过滤服务：加载、热更新与自定义扩展

4.1 启动加载与快速 reload：词库热更新的正确姿势

生产环境里词库不会永远不变，运营每周都会往自定义词库加新词。如果把词库操作直接耦合在 HTTP 请求里，每次建自动机都要几百毫秒，接口必然超时。常见做法是：服务启动时加载一次，之后提供一个 reload 接口，或者用线程定期检查词库文件的修改时间。

import threading from pathlib import Path class FilterService: def __init__(self, word_file: str): self.word_file = word_file self._lock = threading.RLock() self._automaton = self._load_automaton(word_file) def _load_automaton(self, word_file: str): words = load_words_from_files(word_file) # 复用前面清洗后的读取逻辑 return build_automaton(words) def reload(self): new_automaton = self._load_automaton(self.word_file) with self._lock: self._automaton = new_automaton def check(self, text: str): with self._lock: automaton = self._automaton return find_sensitive_words(text, automaton)

逻辑说明：RLock保证同一线程可以重复获取，其他线程在 reload 时不会拿到旧引用正在被替换的半成品。注意check()里先把自动机引用取出来放在局部变量，再退出锁，这样锁只保护引用替换，不保护耗时匹配过程，避免并发请求互相排队。参数说明：word_file建议用绝对路径，服务中通过reload()触发重建；如果想让词库更新无感知，可以另写一个脚本调用 reload 接口，而不是重启进程。

加载时还有一个容易被忽略的点：自动机的构建会占 CPU，如果词库很大，reload 期间的新请求会遇到 CPU 竞争。所以我会把 reload 设计成异步任务，构建完成后通过事件回调替换引用，避免接口调用方阻塞。对于小型团队，也可以接受同步 reload，但至少在日志里记录构建耗时和词条数，方便追查某次接口变慢的原因。

4.2 HTTP 过滤接口与替换策略

有了服务类，接下来就是把它暴露成 HTTP 接口。我用 FastAPI 写一个 JSON 接口，入参是文本和替换字符，返回是否通过、命中的词和替换后的文本。为了避免一次传超大文本拖垮进程，会在接口层限制文本长度。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() filter_service = FilterService("/data/sensitive_words_clean.txt") class FilterRequest(BaseModel): text: str replace_char: str = "*" class FilterResponse(BaseModel): passed: bool hit_words: list[str] filtered_text: str def merge_intervals(hits): if not hits: return [] hits_sorted = sorted([(s, e) for s, e, _ in hits]) merged = [list(hits_sorted[0])] for s, e in hits_sorted[1:]: if s <= merged[-1][1]: merged[-1][1] = max(merged[-1][1], e) else: merged.append([s, e]) return merged @app.post("/filter", response_model=FilterResponse) def filter_text(req: FilterRequest): if len(req.text) > 2000: raise HTTPException(status_code=413, detail="text too long") hits = filter_service.check(req.text) intervals = merge_intervals(hits) filtered_text = req.text for start, end in sorted(intervals, key=lambda x: x[0], reverse=True): filtered_text = filtered_text[:start] + req.replace_char * (end - start + 1) + filtered_text[end + 1:] hit_words = list(set([w for _, _, w in hits])) return FilterResponse( passed=len(intervals) == 0, hit_words=hit_words, filtered_text=filtered_text, )

逻辑说明：merge_intervals把重叠或相邻的命中区间合并，避免替换时后面区间因为前面替换导致错位。替换时倒序处理合并后的区间，因为后面的区间不受前面替换影响。如果正序替换，前面的区间缩短后，后面起点就错位了，这是我最早踩过的坑。参数说明：replace_char默认星号，也可以传空格或者空字符串；413状态码表示长度超限，客户端可以根据状态码提示用户。list[str]语法需要 Python 3.9+，如果你的服务还在 3.8，改成List[str]并from typing import List。

这个接口只做了最简单的同步过滤，真正的生产环境还要考虑：同一用户短时间多次调用需要限流；敏感词命中的结果要不要落库用来分析。这些都可以加，但不属于过滤核心。

4.3 三种扩展方式：自定义词库、正则规则、同音变体

标题里“可以自行扩展”是我认为最关键的一半。某度词库是通用底料，但每个业务的敏感点完全不同：游戏社区可能对“外挂”相关词敏感，金融社区可能要封掉更多理财诈骗话术。因此我会把词库拆成基础词库和业务自定义词库两个文件，构建自动机时合并加载。

def load_words_from_files(*file_paths): words = set() for path in file_paths: with open(path, encoding="utf-8") as f: for line in f: w = line.strip().strip("\ufeff") if w and not w.startswith("#"): words.add(w) return words words = load_words_from_files("sensitive_words_clean.txt", "custom_words.txt") automaton = build_automaton(words)

逻辑说明：合并后的set天然去重，自定义词库中的词如果基础词库已有，不会重复建节点。参数说明：custom_words.txt的格式与基础词库完全一致，运营同学可以直接往文件里加行，不需要改代码。为了避免误操作，我会在部署脚本里对自定义词库做合法性检查，确保每行不是超长 URL、没有异常符号。

再往上是正则维度。自动机只擅长精确串匹配，没办法表达“通配”和“组合”，所以我会额外维护一份正则规则列表，在匹配完自动机后再跑一遍。比如“某X某”这种中间夹一个可变字符的规则，正则可以写成某.某，但要注意正则的性能和误杀率。我的经验：正则规则最多 20 条，且每条都要经过误杀率验证，否则正则的贪婪匹配会在长文本上造成灾难。

扩展方式不止文本文件。常见做法还有：对拼音和同音字做一张改写表，在预处理阶段把文本归一化成拼音后再匹配；也可以维护一份“数字黑话”映射，把常见数字组合映射到对应敏感词位。这些扩展适合业务已经明确出现某类绕过时再加，不建议一开始就全量上，否则你会被各种噪声折腾到怀疑人生。

5. 敏感词过滤避坑指南：5 个让我反复翻车的现场

5.1 误杀正常词：某度敏感词库对语境的盲区

现象：用户说“老板是个好老板”，原文没有违规词，结果被替换成“老板是个**板”。原因是基础词库把“老板”收了进去，某度词库的收录逻辑更偏向网页内容过滤，对日常用语往往没做语境隔离。

解决：我后来在业务层加了两道措施。第一，建立白名单词表，比如“老板”“小姐”这类歧义词，只有在不能命中扩展规则时才放行；第二，把高频误杀词单独拿出来，建立“业务豁免词库”，在匹配前先把白名单词替换成一个较长且不会出现在正常输入里的占位符，如__BW_1__，匹配完成后再还原。注意这属于应急手段，会绕过敏感判断，必须白名单审计可控。

5.2 扩展后内存与构建时间暴涨

现象：自定义词库从 1000 条扩到 8000 条后，reload 一次原本 100ms，变成 1 秒多；内存从 200MB 涨到 1.2GB。

原因：AC 自动机的节点数大致等于所有词条的字符数之和，中文按字符切分后仍然会保留大量重复前缀节点；另外自定义词库如果保留了大量变体词，节点数就会成倍增加。

解决：第一，扩展词库时只加“最小可辨识词根”，不要整句加入；第二，把变体词如大小写变化交给归一化逻辑处理，而不是每个变体都建节点；第三，如果内存实在吃不消，换用对中文更友好的双数组 Trie 实现。我的经验是，某度基础词库加自定义词库加少量正则，内存控制在 500MB 以下是可以接受的，超过就要回来看看是不是塞了太多整句。

5.3 插入符号绕过：归一化与坐标映射

现象：文本“你是个傻 叉”能绕过过滤器，因为原始词库里是“傻叉”，中间多了空格直接失配。

原因：很多绕过手法是在字间插入空格、短横线、零宽字符，AC 自动机匹配的是原始文本，无法跨字符。

解决：在匹配前先做归一化，把预设字符集合内的字符删掉。同时保留一个索引映射，从清洗后文本的位置反推原始文本位置。我用一个数组记录清洗后每个字符对应原文本的下标，匹配完成后去原文本里做替换。这一步不做的话，你会看到替换结果缺字或替换到错误位置。归一化对性能有损耗，建议在进入自动机前只做一次，不要把整个流程拆散到每次命中回调里。

5.4 文件编码不统一导致匹配全失真

现象：从某渠道拿到一份 GBK 编码的词库，脚本用 utf-8 读出来一大片乱码，直接用乱码建自动机，配什么都没有命中。

原因：Windows 环境下很多人工维护的词库是 GBK；如果读取时没有指定编码，默认按系统区域设置，很可能和写入时不一致。

解决：读取文件时先尝试 utf-8，抛异常再用 gbk，再统一转成 utf-8 保存。我的落地脚本里会先检测前 3 个字节是否 BOM，再用编码识别工具兜底；处理完的文件全部改成 utf-8 无 BOM 格式。编码问题看着小，但一旦上线排查起来非常像玄学，因为词表肉眼可见、匹配永远不中，很容易怀疑自动机写错了。

5.5 热更新时出现半新半旧的过滤态

现象：运营修改了自定义词库，调用 reload 后，一部分请求用了新词库，另一部分还是旧词库，表现为同一个违规词时而被拦时而被放。

原因：reload 时直接修改了自动机对象，或者修改了自动机引用的共享列表，导致并发请求踩到不一致状态。

解决：构建一个全新的自动机对象，然后原子替换引用，锁只负责赋值动作。检查代码时，重点看check()方法有没有持有旧自动机的时间过长。只要保证reload()是“先建新，再换引用”，就不会出现半新半旧。另外，在 reload 前先对词库文件做完整性检查，避免只写了一半的文件被加载。

6. 进阶：让敏感词过滤在业务里不那么“憨”

6.1 用回归样本卡住迭代指标

每改一次词库或算法，我都会跑一个固定样本集，包含正常样本和违规变体。样本构成如下表：

跑完只记录三个数：误杀数、漏杀数、无响应超时数。如果误杀比上次多，说明词库扩展失控；如果漏杀比上次少，说明新规则有效。这个回归不需要复杂框架，一个脚本加两个文件就够，但必须坚持下来。我吃过一次亏：上线前临时加了一组正则，结果把“明天见”误杀成“明天*”，回归样本跑出来的时候庆幸多看了一眼。

6.2 白名单与规则审计的配合

白名单是误杀后悔药，但也容易变成绕过后门。我只允许白名单做全词匹配，匹配前先把白名单词替换成占位符，匹配完成后再还原。同时每条白名单都要记录操作人、时间、原因，每周导出一次看看白名单扩容速度。如果某个白名单词出现频率异常高，就要重新评估它的必要性。

我现在每次上线前，还会刻意对过滤器做一次“坏样本攻击测试”：把自己当成绕过者，对每条敏感词尝试加空格、加短横线、换同音字、拼音首字母缩写。这个过程听起来中二，但确实能发现一半以上的规则漏洞。慢慢养成了习惯，我改完词库第一反应不是拿正常文本验证，而是先攻击自己。这套方法不一定适合所有业务，但至少能让敏感词过滤这个黑匣子在你手里透明一点。希望帮到你。

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