简介这是一份面向 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(encodingutf-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, encodingutf-8) as out_f: for line in RAW_FILE.open(r, encodingutf-8, errorsignore): 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))逻辑说明errorsignore会让无法按 utf-8 解码的字符被静默丢弃这在处理半坏文件时比较稳妥但也会丢掉一些生僻字如果你确定词库原本是 GBK更可控的做法是先尝试 utf-8失败后用 gbk 解码再输出成 utf-8。参数说明跳过超过 50 字符的词条是我自己的经验值因为 URL 式噪声既难匹配又会拖慢构建如果你的业务需要匹配长句可以根据词长分布调整这个阈值。清洗脚本里用set去重最终写回 utf-8 纯文本方便后续程序和人工继续维护。清洗之后再跑一次 2.1 的统计脚本你会发现词条数可能缩水三分之一但覆盖率反而更稳。因为重复和大小写变体占了大量冗余不清洗直接建自动机不仅匹配时会命中同一个词多次扩展自定义词库时也不好判断到底加没加过。清洗后的文件建议纳入版本管理每次更新词库时先用户 diff 确认变更量避免某次脚本误删大量词条还不自知。2.3 覆盖率与误杀率用数据说话只有词库没有验证指标等于在裸奔。我会准备两组样本一组是正常的业务语料比如从历史日志里脱敏抽取的聊天记录、评论、昵称一组是人为构造的违规样本。前者用来测误杀率误杀率等于被错判为敏感词的正常消息数除以样本总数后者用来测漏杀率构造方式包括原始词条、大小写变体、符号插入、谐音变体等。指标计算方式合理区间说明误杀率假阳性正常样本中被拦截数量 / 正常样本总数越低越好一般 1%误杀高会让用户骂产品漏杀率假阴性违规样本中未命中数量 / 违规样本总数越小越好但不能只看这个为了漏杀率把词库加满误杀会失控命中分布每个词被命中的次数长尾合理找出哪些词高频误杀、哪些词从未命中构造违规样本时建议在每个原始词条上做几组变体原词、全大写/全小写、中间插入一个空格、中间插入一个短横线、简体繁体互换。这部分样本数量不用多每类几十条即可但必须覆盖“符号插入”这类常见绕过手段。把样本保存成两个明文文件后续每次改词库或改算法后都要重跑一遍形成回归基线。这个习惯能救你很多次——我至少三次因为贪图省事没回归结果上线后误杀率直接翻倍那感觉比挖坑还难受。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_sensitiveFalse时把输入文本和词库统一转为小写再匹配skip_chars设置一个字符集合用于在匹配前把常见的符号和空格替换成空字符用来对抗“插字绕过”max_hit_count控制最多返回多少个命中防止恶意长文本塞进来后 hits 列表膨胀。参数默认值说明注意case_sensitiveFalse英文大小写归一中文不受影响skip_chars ._-#匹配前把这些字符删掉删字符会改变原始偏移替换时要单独处理max_hit_count20单个请求最多返回命中数超出后只保留前 20 个automaton 存储层级字符串pyahocorasick 默认按字符处理不要手动转 bytes中文会切碎设计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_modelFilterResponse) def filter_text(req: FilterRequest): if len(req.text) 2000: raise HTTPException(status_code413, detailtext too long) hits filter_service.check(req.text) intervals merge_intervals(hits) filtered_text req.text for start, end in sorted(intervals, keylambda x: x[0], reverseTrue): 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( passedlen(intervals) 0, hit_wordshit_words, filtered_textfiltered_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, encodingutf-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 用回归样本卡住迭代指标每改一次词库或算法我都会跑一个固定样本集包含正常样本和违规变体。样本构成如下表类型数量内容要求正常样本100覆盖曾经误杀过的词、日常聊天、业务关键词违规样本100覆盖原始词、符号插入、大小写、谐音超长文本20单条 2000 字用来测性能上限跑完只记录三个数误杀数、漏杀数、无响应超时数。如果误杀比上次多说明词库扩展失控如果漏杀比上次少说明新规则有效。这个回归不需要复杂框架一个脚本加两个文件就够但必须坚持下来。我吃过一次亏上线前临时加了一组正则结果把“明天见”误杀成“明天*”回归样本跑出来的时候庆幸多看了一眼。6.2 白名单与规则审计的配合白名单是误杀后悔药但也容易变成绕过后门。我只允许白名单做全词匹配匹配前先把白名单词替换成占位符匹配完成后再还原。同时每条白名单都要记录操作人、时间、原因每周导出一次看看白名单扩容速度。如果某个白名单词出现频率异常高就要重新评估它的必要性。我现在每次上线前还会刻意对过滤器做一次“坏样本攻击测试”把自己当成绕过者对每条敏感词尝试加空格、加短横线、换同音字、拼音首字母缩写。这个过程听起来中二但确实能发现一半以上的规则漏洞。慢慢养成了习惯我改完词库第一反应不是拿正常文本验证而是先攻击自己。这套方法不一定适合所有业务但至少能让敏感词过滤这个黑匣子在你手里透明一点。希望帮到你。本文还有配套的精品资源点击获取
