AC自动机与双数组Trie敏感词检测

可视化演示

通过三种模式展示敏感词检测算法：AC自动机状态机（节点+转移边+失败指针）、双数组Trie压缩存储（base/check表）、文本扫描匹配过程（一次扫描命中多模式），切换模式理解多模式字符串匹配的核心机制。

敏感词检测的挑战

回到我们的图书馆安保场景。你手上的黑名单越来越长：从最初的几十个词，到几千个词，再到几万个词。如果让一个普通管理员逐词查找——拿起第一本书，翻遍全书找“盗窃”；找不到，再翻一遍找“破坏”；再翻一遍找“斗殴”……一本书要翻几十遍，一天下来连一个书架都查不完。这就是多模式匹配（Multi-pattern Matching）的困境：我们需要在一段文本中，同时查找成千上万个关键词。

传统暴力方法的核心问题在于：每次只匹配一个模式，导致文本被反复扫描。想象你在读一页书，每读到一个词，就要和黑名单上的一万个词逐一比对——阅读理解速度瞬间归零。这种暴力扫描的复杂度是O（文本长度 × 关键词总数 × 关键词平均长度），在UGC平台动辄每秒百万条评论的流量下，CPU会直接被打满，这就是高吞吐场景压力。

我们需要一种革命性的思路：让文本只被扫描一遍，就能同时完成所有关键词的匹配。就像图书馆的安保主管换了一种策略——不再逐词查找，而是安排一个“巡逻机器人”，它沿着文本从头走到尾，每走一步就自动检查当前位置是否可能成为某个违规词的开头或中间位置。这，就是AC自动机的雏形。

用比喻回扣：暴力方法是“逐词翻书”，AC自动机是“巡逻机器人一遍走完”。显然，在内容审核的高并发战场上，我们需要后者。理解了挑战，我们来看第一块积木——Trie树。

Trie树基础结构

巡逻机器人不能漫无目的地走，它需要一张“词典地图”。这张地图就是Trie树（前缀树）。想象你把所有敏感词拆成一个个字：“赌博”“赌场”“赌球”——三个词共享前缀“赌”。Trie树的做法是：把共同的字符前缀合并到一个节点，形成一棵以根节点为起点的树。每条从根到某个标记节点的路径，就是一个完整的关键词。

Trie树的插入操作非常直观：从根节点出发，依次检查当前字符是否存在于当前节点的子节点中。若不存在则创建新节点，若存在则沿路径继续。当所有字符插入完毕，将最后一个节点标记为“词尾”（end-of-word）。查找操作同理：从根开始沿着字符边走，若某一步找不到对应子节点，则说明该词不在词典中。

Trie树高效解决了关键词的存储和单模式匹配问题，但它有个明显缺陷：空间浪费。每个节点都要维护一个子节点映射表，对于中文这种大字符集（常用字3000+），每个节点如果开数组存储，会浪费大量空间。更严重的是：Trie树只能做“从头匹配”——如果你想查找“他喜欢她”中的“喜欢”（不从开头开始），Trie树就无能为力了。它像一个只认完整路径的地图，无法告诉你“当前位置是否可以成为某个词的中间部分”。Trie树是基础，但我们需要更强大的自动机——AC自动机。

AC自动机原理

AC自动机（Aho-Corasick Automaton）的核心思想是：把Trie树升级为状态机。什么是状态机？想象那个巡逻机器人身上带了一个“状态指示器”，它每读完一个字符，指示器就跳到一个新状态。这个状态不是简单的“当前字符”，而是“当前文本后缀能匹配到词典中最长的那个前缀”。

怎么理解“状态”？假设词典中有“他”“她”“它”三个词。当机器人读完“他”之后，指示器停在“他”这个状态——代表这已经是一个完整的词。当它继续读到“喜欢”的“喜”时，指示器发现“他喜”不匹配任何前缀，于是自动回退到根状态。但关键来了：如果词典中还有“喜欢”这个词，当机器人读到“喜”时，指示器会跳到“喜”状态；读到“欢”时跳到“喜欢”状态——即便“喜”并不是从文本开头开始的。这正是AC自动机超越Trie树的地方：它能在文本的任意位置开始匹配，而不会遗漏。

用比喻回扣：AC自动机让巡逻机器人具备了“记忆”——它记得自己读到过什么，即使后续字符不匹配，也能快速跳转到“最接近的已匹配状态”，而不是推倒重来。这种“记忆能力”是AC自动机高效的关键。

失败指针设计

失败指针是AC自动机的灵魂。它的设计思想可以概括为一句话：指向当前节点代表路径的最长真后缀。这是什么意思？设当前节点代表的字符串是S（从根到该节点的路径），我们需要找到另一个节点T，使得T代表的字符串是S的真后缀（不包含S自身），并且T是所有满足条件的节点中最长的。用例子说明：假设词典中有“他喜欢”和“喜欢”两个词。当我们匹配到“他喜欢”时，如果下一个字符不匹配“他喜欢”的任何扩展，按照贪心策略，我们希望能退回到“喜欢”这个状态——因为“喜欢”是“他喜欢”的后缀，而且它本身是一个完整词语。

失败指针的构建采用BFS（广度优先搜索）层次遍历：根节点的直接子节点的fail指针指向根节点；对于其他节点，沿着其父节点的fail指针链，查找是否有相同字符的子节点。如果有，则指向该子节点；如果没有，则继续沿fail链向上，直到根节点。这种逐层构建的方式，保证了每个节点的fail指针指向的节点深度一定小于当前节点。

用比喻回扣：失败指针就像巡逻机器人身上的“应急跳转手册”——当它迷路时，不是从头开始，而是按照手册跳到“当前路径最长后缀”的位置，继续巡逻。这保证了匹配过程的线性复杂度——每个字符最多被处理两次（一次正常匹配，一次沿着fail链回退）。整个流程的复杂度分析如下：

双数组Trie压缩

AC自动机解决了匹配效率，但Trie树本身的空间开销在高并发场景下仍然是个问题。每个节点用Map存储子节点，在Java中一个Map对象占用几十字节，百万节点就是几十MB甚至几百MB。想象图书馆的词典地图用了太多纸张，都堆成山了。这时，双数组Trie（Double-Array Trie）登场——它把整棵Trie树压缩成两个数组，空间效率极高。

双数组Trie的核心是两个平行数组：base数组和check数组。base数组记录了每个节点的基准偏移量，用于计算子节点的索引；check数组存储父节点的索引，用于验证匹配的正确性。具体来说：从状态s出发，经过字符c到达状态t，它们满足：

其中code(c)是字符c的编码（如Unicode码点）。这种方法把“Map查子节点”变成了“数组索引+验证”，查询复杂度从哈希查找变为O(1)的数组访问，而且两个数组的空间占用远小于Map结构。

用比喻回扣：双数组Trie把“词典地图”从厚厚的一本册子压缩成两张卡片——一张写着“基准位置”，一张写着“从哪里来”。查找时只需做一次加法和一次比对，速度飞快。在敏感词系统中，AC自动机负责“怎么匹配”，双数组Trie负责“怎么存得快、存得省”，两者结合，天衣无缝。

工程实践与调优

有了核心算法，接下来是落地时的工程难题。首先是中文分词适配。AC自动机天然适用于单字符粒度的匹配，但中文敏感词常常是词语或短语，例如“代开发票”。在匹配时，我们不需要显式分词，只需要把整句文本作为字符流逐一处理即可。但有一个关键细节：全角与半角、繁体与简体的归一化——“代開發票”应该命中“代开发票”。实践中需要对文本做预处理（全角转半角、繁体转简体、去除零宽字符），再做匹配。

变体检测策略是另一个难点。用户会故意用“代开-发-票”“代开F票”等变体绕过过滤。常用策略包括：在预处理阶段删除所有非字母数字字符（但保留中文字符），再进行匹配；或者建立“变体映射表”，将同音字、形近字映射为规范字。比如“艹”映射到“草”，“淦”映射到“干”。这本质上是在预处理和分词阶段做文章，为AC自动机提供“干净”的输入。

并发性能优化方面，对于每秒百万级请求的高吞吐场景，常用的优化有：

1. 预编译自动机：在系统启动时将敏感词库构建为AC自动机，运行时只读不写，避免加锁

2. 线程池+无状态匹配：AC自动机的匹配过程只依赖输入文本，不修改自动机状态，因此天然线程安全，可以放心并发

3. 批处理：将多条短文本合并成一条长文本批量匹配，减少函数调用开销

4. 内存池：双数组Trie预先分配固定大小的数组，避免运行期扩容

用比喻回扣：我们的图书馆安保系统终于建成——预处理是“安检门”（清洗文本），变体映射是“化妆识别”（归一化），AC自动机是“巡逻机器人”（高效匹配），双数组Trie是“极简词典”（极致存储）。这套组合拳，能扛住任何高并发的文本审核需求。

---

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""AC自动机与双数组Trie敏感词检测示例"""from collections import deque  class ACTrieNode:    """AC自动机节点"""    __slots__ = ('children', 'fail', 'is_end')     def __init__(self):        self.children = {}        self.fail = None       # 失败指针        self.is_end = False    # 是否为词尾  class ACTrie:    """AC自动机"""     def __init__(self, words):        self.root = ACTrieNode()        for word in words:            # 使用规范化后的词构建Trie树            self._insert(normalize(word))        self._build_fail()     def _insert(self, word):        node = self.root        for ch in word:            if ch not in node.children:                node.children[ch] = ACTrieNode()            node = node.children[ch]        node.is_end = True     def _build_fail(self):        """BFS构建失败指针"""        q = deque()        for child in self.root.children.values():            child.fail = self.root            q.append(child)        while q:            cur = q.popleft()            for ch, child in cur.children.items():                f = cur.fail                while f is not None and ch not in f.children:                    f = f.fail                child.fail = f.children[ch] if f else self.root                if child.fail:                    child.is_end = child.is_end or child.fail.is_end                q.append(child)     def search(self, text):        """一次扫描，返回所有命中的敏感词"""        results = []        node = self.root        for i, ch in enumerate(text):            while node is not None and ch not in node.children:                node = node.fail            if node is None:                node = self.root                continue            node = node.children[ch]            tmp = node            while tmp is not None:                if tmp.is_end:                    # 通过回溯找到完整词                    words = self._collect_words(tmp)                    for w in words:                        start = i - len(w) + 1                        results.append((start, i, w))                tmp = tmp.fail        return results     def _collect_words(self, node):        """从节点回溯收集所有命中词（简化版）"""        words = []        current = node        chars = []        while current is not None and current is not self.root:            # 简化：仅返回最长词，实际应用需记录路径            words.append('')            current = current.fail        return words  def normalize(text):    """预处理：大写转小写、全角转半角"""    result = []    for ch in text:        code = ord(ch)        if ch == '　':            result.append(' ')        elif 0xFF01 <= code <= 0xFF5E:            # 全角转半角            result.append(chr(code - 0xFEE0))        else:            result.append(ch.lower())    return ''.join(result)  def build_double_array(trie):    """将AC自动机的Trie树转换为双数组Trie（示意）"""    base = [0]    check = [0]    node_index = {}    next_index = 1     def dfs(node, idx):        if node.is_end:            check[idx] = -idx  # 词尾标记        for ch, child in node.children.items():            code = ord(ch)            target_idx = idx + code            # 冲突处理：向后探测空位            while target_idx < len(base) and check[target_idx] != 0:                target_idx += 1            while len(base) <= target_idx:                base.append(0)                check.append(0)            base[target_idx] = 0            check[target_idx] = idx            node_index[id(child)] = target_idx            dfs(child, target_idx)     dfs(trie.root, 0)    return base, check  if __name__ == '__main__':    # 敏感词库    words = ['赌博', '赌场', '发票', '代开发票']    ac = ACTrie(words)    test_text = '请加微信代开发票，各种赌博网站均可接待，发票代开请联系'    print('测试文本:', test_text)    print('命中结果:', ac.search(normalize(test_text)))    # 双数组Trie构建示意    base_arr, check_arr = build_double_array(ac)    print('双数组大小: base =', len(base_arr), 'check =', len(check_arr))

1. child.fail = f.children[ch] if f else self.root沿失败链查找具有相同字符子节点的节点，找到则指向该子节点
2. child.is_end = child.is_end or child.fail.is_end如果失败指针指向的节点是词尾，当前节点也应标记为词尾（继承性）
3. while node is not None and ch not in node.children: node = node.fail匹配失败时沿fail链回退
4. code = ord(ch)获取字符的Unicode编码，作为双数组中计算偏移量的依据
5. target_idx = idx + code核心的双数组索引公式：子节点位置=父节点base+字符编码
6. check[target_idx] = idx记录父节点索引，用于验证转移的合法性

常见误区

面试题