AC 自动机

AC 自动机（Aho–Corasick）是一种多模式字符串匹配算法，能在一段文本中同时查找多个关键词，并返回它们的出现位置。 它把词库构建成一组匹配状态和转移规则，读入文本字符时按规则切换状态，因此称为“自动机”。同一份词库构建完成后，可以反复搜索不同文本。

1. AC 自动机的用途

• 敏感词筛查：按词库检查评论、昵称或消息，将命中词和位置交给审核规则。
• 术语提取：按词典查找产品名、机构名或专业术语，关联对应的业务 ID。
• 日志扫描：同时查找多种错误短语或事件标记，根据命中位置提取附近的日志内容。
• 关键词高亮与替换：在页面上标出命中的关键词，或按业务规则替换为指定文本。

要查找的关键词叫作模式串，被搜索的字符串叫作文本。

词库包含“苹果”“红苹果”“苹果汁”“橙汁”，文本是“一杯红苹果汁”，能够找到“红苹果”“苹果”“苹果汁”。

全文的位置统一使用从 0 开始、左闭右开的区间 [start, end)，与 Python 的 text[start:end] 对应。

即使命中位置有重叠，每个关键词的匹配结果也都会保留。 只取最长词、排除重叠等处理由业务规则决定。

AC 匹配的是连续出现的相同字符。 同义词和拼写纠错需要额外规则或其他算法。

“苹果”会命中“红苹果汁”中间的两个字；“橙汁”不会匹配“橙子汁”，因为中间多了一个“子”。

如果每个关键词都单独查找，同一段文本就要被反复扫描。AC 把这些查找合在一起，每读一个字符就更新当前匹配状态，文本下标不用回退。

自动机要花时间构建，也会占用内存。选择时要看词库大小，以及它会被用多少次：

• 词库相对稳定，要搜索多段文本：构建一次，就能反复使用。
• 关键词很少，或只搜索一次：直接调用字符串查找函数通常更省事。哪种更快，要用实际词库和文本比较。

2. 工作原理

2.1 Trie 保存共同前缀

Trie（字典树）是一种保存和查找字符串的树形数据结构。 它把关键词按字符逐层存储，开头相同的词共用同一段路径。

词开头连续的一段叫作前缀，例如“苹”“苹果”“苹果汁”都是“苹果汁”的前缀。

根节点是所有词的共同起点，代表空字符串。连接节点的边记录一个字符；从根节点沿边向下走，经过的字符依次拼起来，就得到当前节点代表的前缀。

向 Trie 加入“苹果”时，建立经过“苹”和“果”两条边的路径，在末尾标记这个词结束。加入“苹果汁”时，已有的“苹果”路径可以复用，只需增加一条“汁”边，并标记新的词尾。“红苹果”和“橙汁”的开头不同，各自从根节点建立分支。

括号中标出整条路径，* 表示一个完整关键词在这里结束。

1
2
3
4

root
├─ 苹 → (苹) ─ 果 → (苹果)* ─ 汁 → (苹果汁)*
├─ 红 → (红) ─ 苹 → (红苹) ─ 果 → (红苹果)*
└─ 橙 → (橙) ─ 汁 → (橙汁)*

“红苹果”末尾的“苹果”与根节点下的“苹果”属于两条路径，Trie 只共享词开头相同的部分。

查找“苹果”时，从根节点依次沿“苹”“果”边走，末尾有结束标记，说明词库中存在这个词。查找“苹”虽然能走到对应节点，却没有结束标记，说明它只是前缀，不是词库中的完整词。

“苹果”是关键词结束节点，同时还有通向“苹果汁”的子节点。

关键词结束不等于搜索结束，后面的字符仍可能组成更长的关键词。

仅有 Trie 时，可以从某个文本起点沿字符边尝试匹配。断路后若换一个起点重新比较，已经读过的字符还会被重复检查。AC 为节点增加失败指针，保存断路后仍有用的匹配信息。

2.2 失败指针保留可继续匹配的后缀

扫描“一杯红苹果汁”，读完“红苹果”后，接下来是“汁”。“红苹果”节点没有“汁”这条边，但刚读过的后两个字符“苹果”，恰好是关键词“苹果汁”的前缀。因此可以把当前状态改成“苹果”，接上“汁”，找到“苹果汁”。

1
2
3
4

已读文本：一杯红苹果
待处理字符：汁

(红苹果) ─ fail → (苹果) ─ 汁 → (苹果汁)

字符串末尾连续的一段叫作后缀；排除字符串自身，剩下的后缀叫作真后缀。“红苹果”的真后缀有“苹果”“果”和空字符串，其中“苹果”是另一个词的前缀，可以接着匹配。

失败指针 fail 指向的节点，代表当前路径最长的、仍是某个关键词前缀的真后缀。 它不一定指向父节点；找不到非空候选时，指向根节点。

例如，fail(红苹果) = 苹果，即从“红苹果”跳到另一条分支上的“苹果”节点。

跳转只改变当前状态，不移动文本下标，也没有消耗待处理的“汁”。 如果候选节点仍接不上当前字符，就沿它的 fail 继续找；退到根节点仍没有对应边时，这个字符处理结束，状态留在根节点。

每处理完一个字符，当前状态表示：已读文本的后缀中，能够在 Trie 中找到的最长前缀路径。

2.3 同一位置可能结束多个关键词

读到“红苹果”的末尾时，“红苹果”和“苹果”都已经完整出现。如果只检查当前节点的结束标记，就会漏掉“苹果”。

沿失败链能够找到这些后缀，但每读一个字符都遍历整条链，会检查很多并非关键词结尾的节点。这里增加输出链接 output：指向失败链上最近的关键词结束节点；不存在时记为 None。

output(红苹果) = 苹果。输出“红苹果”后，沿输出链接找到“苹果”，一并输出。遇到 None 就结束。

结果收集使用单独的指针，扫描状态仍留在“红苹果”，以便处理下一个字符。

表中的状态名对应上图的路径，不是已经输出的最后一个词。

3. Python 实现

3.1 按层构建失败指针

每个节点保存四项信息：

• children：记录读到某个字符时，能走到哪个子节点。
• fail：当前字符接不上时，跳到哪个节点继续尝试。
• word：记录在这个节点结束的完整关键词。
• output：指向失败链上最近的关键词结束节点。

根节点的 fail 指向自身。第一层只有一个字符，没有非空真后缀，fail 都指向根节点。其余节点通过 广度优先遍历（BFS） 构建，也就是按深度逐层处理。

设父节点为 parent，字符边 char 指向孩子 child：

1. 候选位置取 parent.fail。它代表父路径可以保留的最长真后缀。
2. 候选位置没有 char 边时，沿候选位置的失败链继续查找。
3. 找到 char 边，就把 child.fail 指向这条边的终点；根节点也接不上时，指向根节点。
4. 若 child.fail 是关键词结束节点，child.output 就指向它；否则沿用它的 output。

构建“红苹”节点的失败指针时，它的父节点是“红”，fail(红) 是根节点。根节点存在“苹”边，所以 fail(红苹) = 苹。构建“红苹果”节点时，fail(红苹) = 苹，“苹”节点存在“果”边，因此 fail(红苹果) = 苹果。

除根节点外，失败指针总指向更浅的节点。 BFS 处理某个孩子时，它所依赖的失败指针和输出链接已经建立，不需要重新推导整段后缀。

3.2 完整代码

输入词库中的词为非空、互不重复的 str，匹配区分大小写。构造对象时完成建树；finditer(text) 逐个返回 (关键词, start, end)，保留全部重叠匹配。

 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

"""AC 自动机教学实现；模式串为非空、互不重复的 str，位置使用左闭右开的字符下标。"""

from __future__ import annotations

from collections import deque
from collections.abc import Iterable, Iterator


class Node:
    def __init__(self):
        self.children: dict[str, Node] = {}
        self.fail: Node = self
        self.output: Node | None = None
        self.word: str | None = None


class AhoCorasick:
    def __init__(self, words: Iterable[str]):
        self.root = Node()
        root = self.root
        for word in words:
            node = root
            for char in word:
                if char not in node.children:
                    node.children[char] = Node()
                node = node.children[char]
            node.word = word

        # 根节点的 fail 指向自身；第一层节点没有非空真后缀，fail 都指向根节点。
        queue = deque(root.children.values())
        for node in queue:
            node.fail = root

        while queue:
            parent = queue.popleft()
            for char, child in parent.children.items():
                candidate = parent.fail
                while candidate is not root and char not in candidate.children:
                    candidate = candidate.fail
                child.fail = candidate.children.get(char, root)
                # BFS 保证较浅节点已处理；output 只连接失败链上最近的关键词结束节点。
                child.output = child.fail if child.fail.word is not None else child.fail.output
                queue.append(child)

    def finditer(self, text: str) -> Iterator[tuple[str, int, int]]:
        """逐个产生 (关键词, 起始下标, 结束下标)，保留所有重叠匹配。"""
        root = self.root
        state = root
        for end, char in enumerate(text, start=1):
            # 失败跳转不消耗 char；找到能接上它的节点，或退回根节点。
            while state is not root and char not in state.children:
                state = state.fail
            state = state.children.get(char, root)

            # 报告当前状态及其后缀中的关键词，报告过程不改变扫描状态。
            match: Node | None = state
            while match is not None:
                if match.word is not None:
                    yield match.word, end - len(match.word), end
                match = match.output


if __name__ == "__main__":
    automaton = AhoCorasick(["苹果", "红苹果", "苹果汁", "橙汁"])
    for word, start, end in automaton.finditer("一杯红苹果汁"):
        print(f"{word}[{start}:{end}]")

1
2
3

# 红苹果[2:5]
# 苹果[3:5]
# 苹果汁[3:6]

finditer 按结束位置从左到右返回匹配结果。同一结束位置，匹配到的词按长度从长到短返回。因此结果不一定按起始位置排序。每次搜索新文本都从根节点开始。

3.3 时间与空间成本

设词库总字符数为 M，Trie 节点数为 S，文本长度为 n，返回的匹配数为 z。这里按 Python 字典查询平均为 O(1) 计算。

• 构建时间：O(M)。 插入词库需要遍历 M 个字符。

  构建失败指针时，沿一个关键词的路径向下走，每经过一条字符边，候选后缀的深度最多增加 1；沿失败指针回退时，深度会减小。因此，沿这条路径回退的总层数不会超过词长。

  把所有关键词路径的计算量加起来，上限是 O(M)。共享的边实际只处理一次。

• 搜索时间：O(n + z)。 沿字符边走一步，状态深度最多增加 1；每次失败跳转都会让深度减小。所以，扫描 n 个字符期间，失败跳转的总次数是 O(n)。

  输出链接只访问关键词结束节点，每访问一个就输出一个匹配。返回 z 个匹配需要额外的 O(z) 时间。

• 自动机存储：O(M)。 字符边、失败指针和输出链接占 O(S)，保留的关键词总长为 M，且 S ≤ M + 1。

  构建时，BFS 队列最多占 O(S)。搜索时，迭代器只维护少量状态；如果把全部结果存成列表，还需要 O(z) 空间。

这份实现只保存实际存在的字符边，没有给每个节点分配整个字符集的转移表，也没有把后缀关键词列表复制到每个节点。

4. 生产中的使用

4.1 关键词关联业务数据

Python 项目可以使用 C 扩展库 pyahocorasick。

安装包名为 pyahocorasick，导入名为 ahocorasick。

1

python -m pip install "pyahocorasick==2.2.0"

术语词典包含四条业务记录，其中“北京”同时属于两个条目。

add_word() 对同一个词重复赋值会替换旧值，因此按词聚合关联记录，每个词只添加一次。

 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

import ahocorasick

terms = [
    {"id": 101, "word": "北京", "category": "地名"},
    {"id": 102, "word": "北京大学", "category": "高校"},
    {"id": 103, "word": "大学", "category": "机构类型"},
    {"id": 104, "word": "北京", "category": "主题标签"},
]

# 同一个词可能对应多个业务条目，聚合后只向词库添加一次，避免覆盖旧记录。
records_by_word = {}
for term in terms:
    records_by_word.setdefault(term["word"], []).append((term["id"], term["category"]))

automaton = ahocorasick.Automaton()  # 当前是空的 Trie，可以添加关键词。
for word, records in records_by_word.items():
    automaton.add_word(word, (word, records))  # 保存词本身和业务记录，命中时一起返回。
automaton.make_automaton()  # 根据词库构建失败关系，将 Trie 转成可搜索的 AC 自动机。

text = "我在北京大学读书。"
# iter() 返回所有命中，包括重叠词；每项是末字符下标和 add_word() 存入的值。
for end_index, (word, records) in automaton.iter(text):
    end = end_index + 1  # 库返回包含末字符的下标，转换成 Python 切片的右边界。
    start = end - len(word)
    for term_id, category in records:  # 把一次字符串命中展开为对应的业务条目。
        print(term_id, category, f"[{start}, {end})", text[start:end])

1
2
3
4

# 101 地名 [2, 4) 北京
# 104 主题标签 [2, 4) 北京
# 102 高校 [2, 6) 北京大学
# 103 机构类型 [4, 6) 大学

一次“北京”的字符串命中对应两条业务记录；“北京大学”和“大学”的重叠命中也被保留。

自动机负责找位置，业务 ID 和类别是应用存入的值。

4.2 构建复用与保存加载

服务启动时加载词库并构建自动机，处理请求时复用已经构建好的对象。 每个请求都重新构建，就会重复插入词库、计算失败指针。

自动机可与词库版本、文本处理规则一起序列化保存。

1
2
3
4
5
6
7

import pickle

# 词库版本和文本处理规则随自动机一起保存，加载时可以核对。
bundle = {"dictionary_version": "terms-v1", "normalization": "identity", "automaton": automaton}
with open("terms-v1.pkl", "wb") as file:
    pickle.dump(bundle, file, protocol=pickle.HIGHEST_PROTOCOL)  # 序列化到二进制文件。
print(bundle["dictionary_version"], len(automaton))

1

# terms-v1 3

3 是不同关键词的数量，四条业务记录合并成了三个词；identity 是示例自定义的规则名，表示保留原文本。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pickle

with open("terms-v1.pkl", "rb") as file:
    loaded = pickle.load(file)  # 恢复已构建的自动机，文件必须来自可信的构建流程。

loaded_automaton = loaded["automaton"]  # 直接搜索，无需重新添加词库或构建。
text = "我在北京大学读书。"
print(loaded["dictionary_version"], loaded["normalization"])
for end_index, (word, records) in loaded_automaton.iter(text):
    end = end_index + 1  # 转换为右开区间，起点由关键词长度确定。
    print(word, end - len(word), end)

1
2
3
4

# terms-v1 identity
# 北京 2 4
# 北京大学 2 6
# 大学 4 6

只加载由自己的构建流程生成、来源可确认的 pickle 文件。 反序列化可能执行文件中的代码，不应直接接收用户上传的索引文件。

词库源数据仍需保存，升级 Python、库版本或构建环境时，应验证旧文件能否加载及结果是否一致；必要时由源词库重新构建。

4.3 词库更新与版本切换

增加或删除关键词可能改变失败指针和输出关系。对于这类固定词库服务，建议按版本构建新的自动机，而不是在处理请求的对象上直接修改词库。

1. 根据新版词库构建新对象，核对词条数量、代表文本的命中结果及文本处理规则。
2. 将新对象与新版业务数据一起交给服务；新请求使用新版，正在执行的请求继续持有旧版。
3. 旧请求结束后释放旧对象。

多进程服务要在各个进程中分别加载和切换，并在日志中记录每个请求使用的词库版本。

版本切换时，新旧自动机可能同时占用内存。构建时的临时数据也会占用内存，所以不能只看索引文件的大小。

词库更新频繁时，还要考虑自动机的重建耗时和更新频率。

5. 使用注意事项

• 明确保留哪些匹配。 保留全部匹配、按最左起点选最长词、按业务优先级排除重叠，这几种规则会得到不同的结果。最早返回的词不一定最长。

  “北京大学”会同时命中“北京”“北京大学”“大学”。

  替换文本时，根据原文坐标选出互不重叠的区间，并从右向左替换，避免文本长度变化导致后续位置错位。

• 词库和文本采用相同的处理规则。 大小写转换、全半角转换要同时用于词库和待匹配文本。

  如果转换改变了字符串长度，返回的位置就不能直接用于原文。需要高亮或替换时，记录处理后的位置对应原文的哪个区间。

  示例返回的是 Python 字符下标，不是 UTF-8 字节偏移。

• 匹配越多，处理结果的开销越大。 O(n + z) 中的 z 是匹配数量。重叠命中很多时，遍历结果、展开业务记录和保存结果都会增加开销。

  只判断有没有关键词时，找到一个就能停止。需要全部位置时，就要把处理结果的时间和内存一起算进去。