跳转至

为什么大厂都用 DFA 做敏感词过滤？Python 带你手撸一个！

敏感词过滤，这个看起来“老掉牙”的功能，其实藏着不少算法的门道。
你可能不知道，大厂评论系统、弹幕平台、甚至聊天机器人背后，都在悄悄跑着一台“小型自动机”——DFA。

今天我们就用 Python，带你从最简单的思路出发，一步步搞懂：
为什么大家都爱用 DFA 算法 来做敏感词过滤。

一、为什么要做敏感词过滤？

想象一下，你在一个社交平台写评论，刚发出去就被提示：

“您的内容包含敏感词，请修改后再试。”

是不是立刻联想到：这背后一定有个“词库”在帮忙拦截？
没错，敏感词过滤的核心工作就是在海量文本中快速发现违规词汇。

看似简单的需求，背后其实有点算法味道：

  • 一句话可能几百上千个字；

  • 词库可能上万条；

  • 还要支持中英文混合、模糊匹配……

如果算法写得不高效，系统性能就会大打折扣。
所以，敏感词过滤其实是“性能挑战” + “算法设计”的综合问题。

二、常见的传统过滤方法

在介绍高大上的DFA算法之前，我们先来看看那些“前辈们”是怎么做的。

1. 简单替换法：最直白的“笨办法”

这是最直观的方法，就像用查找替换功能一样简单：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def simple_filter(text, sensitive_words):
    for word in sensitive_words:
        text = text.replace(word, "***")
    return text

# 测试一下
keywords = ("暴力", "色情", "赌博")
content = "这是一段包含暴力和赌博内容的文本。"
print(simple_filter(content, keywords))
# 输出：这是一段包含***和***内容的文本。

👍 优点：

  • 代码超级简单，几行搞定
  • 容易理解和上手，适合初学者

👎 缺点：

  • 性能极低：每个敏感词都要全文扫描一次
  • 如果有1000个敏感词和10000字的文本，理论上需要**1000万次比较**！
  • 无法处理变形词（比如用符号隔开的敏感词）

2. 正则表达式法：文本处理的“瑞士军刀”

正则表达式可以一次性匹配所有敏感词，效率提升明显：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import re

def regex_filter(text, sensitive_words):
    pattern = '|'.join(sensitive_words)
    return re.sub(pattern, '***', text)

# 同样的测试
keywords = ["暴力", "色情", "赌博"]
content = "这是一段包含暴力和赌博内容的文本。"
print(regex_filter(content, keywords))

👍 优点：

  • 比简单替换快一倍多（测试显示：0.48秒 vs 1.24秒）
  • 一行代码完成复杂匹配
  • 支持灵活的匹配规则

👎 缺点：

  • 敏感词多了之后，正则模式变得复杂
  • 可能引发**回溯陷阱**，导致CPU飙升
  • 仍然需要遍历整个敏感词库

3. 传统方法的根本问题

无论简单替换还是正则表达式，都存在一个**致命缺陷**：它们都需要**遍历整个敏感词库**进行匹配。

用专业术语来说，对于包含N个敏感词的词库和长度为M的文本，传统方法的时间复杂度是**O(N×M)**。这意味着数据量稍微大一点，性能就会断崖式下跌！

4. 为什么需要更好的方案？

想象一下真实场景：一个社交平台可能有：

  • 10万+的敏感词库
  • 每秒数千条的文本流量
  • 毫秒级的响应要求

在这种压力下，传统方法根本扛不住！这就是为什么我们需要更智能的算法——DFA算法。

下一节，我们将揭秘这个“性能怪兽”DFA算法，看看它是如何实现质的飞跃的！

三、DFA 算法原理浅析

DFA（Deterministic Finite Automaton，确定性有限自动机）听起来很高大上，可能有点抽象。
别慌，我们来用生活化的方式理解。

想象你走进一个自动门系统：

  • 你每走一步（读一个字符），系统就判断你该去哪一扇门（状态转移）；

  • 一旦你走到“终点房间”（终止状态），说明命中了敏感词。

简单吧？这其实就是 DFA 的运行逻辑。

DFA 状态转移流程图（文字版）

假设我们的敏感词库是：

["枪", "毒品", "恐怖"]

状态机可以想象成这样（每个节点代表匹配的进度）：

1
2
3
4
Start
 ├─ 枪 → [End]
 ├─ 毒 → 品 → [End]
 └─ 恐 → 怖 → [End]

当程序读取文本时，比如遇到“毒”：

  • 系统进入“毒”状态；

  • 下一个字如果是“品”，进入“终止状态”，说明命中“毒品”；

  • 否则回到 Start 状态，继续下一轮。

这就是“有限状态机”在敏感词过滤中的全部精髓。

举个生动例子：假设我们的敏感词有“王八”和“王八蛋”，DFA会构建这样的结构：

1
2
3
4
5
6
7
8
{
    "王": {
        "八": {
            "is_end": True,  # 匹配到“王八”
            "蛋": {"is_end": True}  # 匹配到“王八蛋”
        }
    }
}

算法只需 扫描一遍文本（O(n))，就能完成匹配，性能暴涨！

DFA的“超能力”在哪里？

1. 时间复杂度与敏感词数量无关 这是DFA最厉害的地方！无论你的敏感词库有100个还是10万个词，检测一段文本所需的时间**只取决于文本本身的长度**。对比一下：

算法类型 时间复杂度 10万词库下的表现
传统方法 O(N×M) 随着词库增大线性变慢
DFA算法 O(N) 性能稳定，不受词库大小影响

2. 一次扫描，多词匹配 DFA对待检测文本只需**从头到尾扫描一次**，就能同时检测出所有预设的敏感词。比如检测“匹配关键词”这句话时，DFA会依次转移：匹 → 配（记录“匹配”）→ 关 → 键 → 词（记录“匹配关键词”），一次扫描就找出所有嵌套关键词！

3. 前缀共享，节省空间 具有相同前缀的敏感词会共享树节点，比如“匹配”和“匹配关键词”共享“匹配”这两个字的前缀路径，避免了重复存储。

实际应用中的选择建议

  • DFA适用场景：非常适合**敏感词库大、性能要求高**的在线服务，如社交媒体评论、实时聊天、内容审核系统等
  • DFA的代价：典型的“空间换时间”，构建好的状态机可能占用较多内存
  • 工业级优化：实际应用中常使用**双数组Trie树**，在保留DFA高效匹配优点的同时，能显著减少内存占用

性能数据说话：在10000个敏感词的场景下，DFA算法处理1MB文本仅需**0.12秒**，而简单替换法需要1.24秒，正则表达式法需要0.48秒。当词库规模达到10万+时，这种优势会更加明显！

现在你已经了解了DFA算法的核心优势，接下来就让我们进入最激动人心的部分——手把手用Python实现这个“加速器”！

三、手把手教你用Python实现DFA敏感词过滤

前面我们已经了解了传统方法的局限性和DFA算法的优势，现在让我们亲自动手，用Python实现一个完整的DFA敏感词过滤器！

第一步：创建DFA过滤器类

我们先来搭建基础的框架：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# -*- coding:utf-8 -*-
class DFAFilter:
    def __init__(self):
        self.keyword_chains = {}  # 关键词链表（字典树）
        self.delimit = '\x00'     # 结束标记符

    def add(self, keyword):
        """添加敏感词到字典树"""
        keyword = keyword.lower().strip()  # 转为小写并去除空格
        if not keyword:
            return

        level = self.keyword_chains
        # 遍历敏感词的每个字符
        for i in range(len(keyword)):
            char = keyword[i]
            # 如果字符已存在，进入下一层
            if char in level:
                level = level[char]
            else:
                # 如果字符不存在，创建新节点
                if not isinstance(level, dict):
                    break
                for j in range(i, len(keyword)):
                    level[keyword[j]] = {}
                    last_level, last_char = level, keyword[j]
                    level = level[keyword[j]]
                # 标记敏感词结束
                last_level[last_char] = {self.delimit: 0}
                break

        # 处理到达词尾的情况
        if i == len(keyword) - 1:
            level[self.delimit] = 0

代码解析：

  • __init__方法初始化了敏感词树和结束标记
  • add方法负责将敏感词逐个字符添加到树结构中
  • 通过字典嵌套的方式构建树形结构，共享相同前缀

第二步：批量添加敏感词和文件读取

实际应用中，我们需要从文件加载敏感词库：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
def parse(self, keywords):
    """批量添加敏感词"""
    for keyword in keywords:
        self.add(keyword)

def parse_from_file(self, filepath):
    """从文件读取敏感词库"""
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                self.add(line.strip())
        print(f"敏感词库加载完成")
    except FileNotFoundError:
        print(f"错误：找不到文件 {filepath}")

第三步：核心过滤功能实现

这是整个DFA过滤器的核心部分：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def filter(self, text, repl="*"):
    """过滤文本中的敏感词"""
    text_lower = text.lower()  # 转为小写进行匹配
    result = []  # 保存过滤结果
    start = 0    # 检测起始位置

    while start < len(text_lower):
        level = self.keyword_chains
        step_ins = 0  # 匹配到的字符数

        # 从start位置开始检测
        for char in text_lower[start:]:
            if char in level:
                step_ins += 1
                # 如果到达敏感词结尾
                if self.delimit not in level[char]:
                    level = level[char]
                else:
                    # 替换敏感词
                    result.append(repl * step_ins)
                    start += step_ins - 1
                    break
            else:
                # 当前字符不匹配，保留原字符
                result.append(text[start])
                break
        else:
            result.append(text[start])

        start += 1

    return ''.join(result)

第四步：完整可运行的示例

让我们把所有的代码整合起来，创建一个可以直接运行的完整示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
# -*- coding:utf-8 -*-
import time

class DFAFilter:
    def __init__(self):
        self.keyword_chains = {}
        self.delimit = '\x00'

    def add(self, keyword):
        keyword = keyword.lower().strip()
        if not keyword:
            return

        level = self.keyword_chains
        for i in range(len(keyword)):
            char = keyword[i]
            if char in level:
                level = level[char]
            else:
                if not isinstance(level, dict):
                    break
                for j in range(i, len(keyword)):
                    level[keyword[j]] = {}
                    last_level, last_char = level, keyword[j]
                    level = level[keyword[j]]
                last_level[last_char] = {self.delimit: 0}
                break

        if i == len(keyword) - 1:
            level[self.delimit] = 0

    def parse(self, keywords):
        for keyword in keywords:
            self.add(keyword)

    def filter(self, text, repl="*"):
        text_lower = text.lower()
        result = []
        start = 0

        while start < len(text_lower):
            level = self.keyword_chains
            step_ins = 0

            for char in text_lower[start:]:
                if char in level:
                    step_ins += 1
                    if self.delimit not in level[char]:
                        level = level[char]
                    else:
                        result.append(repl * step_ins)
                        start += step_ins - 1
                        break
                else:
                    result.append(text[start])
                    break
            else:
                result.append(text[start])
            start += 1

        return ''.join(result)

# 使用示例
if __name__ == "__main__":
    # 创建过滤器
    dfa = DFAFilter()

    # 添加敏感词
    sensitive_words = ["傻逼", "傻子", "坏蛋", "坏人", "垃圾"]
    dfa.parse(sensitive_words)

    # 测试文本
    test_text = "你真是个傻逼，坏蛋！这个产品质量太垃圾了。"

    # 过滤文本
    start_time = time.time()
    filtered_text = dfa.filter(test_text)
    end_time = time.time()

    print("原始文本：", test_text)
    print("过滤后：", filtered_text)
    print(f"处理时间：{end_time - start_time:.6f}秒")

运行结果示例：

1
2
3
原始文本： 你真是个傻逼，坏蛋！这个产品质量太垃圾了。
过滤后： 你真是个**，**！这个产品质量太**了。
处理时间：0.000156秒

进阶功能：处理特殊情况

在实际应用中，我们还需要考虑一些边界情况：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def enhanced_filter(self, text, repl="*", skip_chars=None):
    """增强版过滤器，可跳过特定字符"""
    if skip_chars is None:
        skip_chars = [' ', '!', '！', '@', '#', '$', '%', '?', '？']

    text_lower = text.lower()
    result = []
    start = 0

    while start < len(text_lower):
        # 跳过无意义字符
        if text_lower[start] in skip_chars:
            result.append(text[start])
            start += 1
            continue

        level = self.keyword_chains
        step_ins = 0
        match_positions = []

        # 检测敏感词（考虑跳过无意义字符）
        pos = start
        while pos < len(text_lower):
            char = text_lower[pos]
            if char in skip_chars:
                pos += 1
                continue

            if char in level:
                match_positions.append(pos)
                step_ins += 1
                if self.delimit not in level[char]:
                    level = level[char]
                else:
                    # 找到敏感词，进行替换
                    result.append(repl * step_ins)
                    start = pos
                    break
                pos += 1
            else:
                break

        start += 1

    return ''.join(result)

实用技巧：持久化DFA树

对于生产环境，我们可以将构建好的DFA树保存到文件，避免每次重启都重新构建：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import pickle

def save_dfa_tree(dfa, filename):
    """保存DFA树到文件"""
    with open(filename, 'wb') as f:
        pickle.dump(dfa.keyword_chains, f)

def load_dfa_tree(dfa, filename):
    """从文件加载DFA树"""
    with open(filename, 'rb') as f:
        dfa.keyword_chains = pickle.load(f)

# 使用示例
dfa = DFAFilter()
dfa.parse(["敏感词1", "敏感词2"])  # 构建DFA树
save_dfa_tree(dfa, "dfa_tree.pkl")  # 保存到文件

# 下次启动时直接加载
dfa2 = DFAFilter()
load_dfa_tree(dfa2, "dfa_tree.pkl")  # 从文件加载

通过这个完整的教程，你已经掌握了用Python实现DFA敏感词过滤的核心技术。从基础实现到性能优化，再到生产环境的实用技巧，你现在可以自信地在自己的项目中应用这个强大的算法了！

记住，DFA算法的精髓在于**用空间换时间**，通过预处理构建高效的查询结构，实现快速匹配。这种思想在很多算法设计中都有应用，值得深入理解和掌握。