跳轉到

10.1 敏感詞過濾系統#

概念說明#

論壇、聊天、電商評論都需要敏感詞過濾:即時判斷文字是否含有禁用詞並阻擋或替換。挑戰有三層:

  1. 規模:詞庫常達 5 萬~50 萬條(含變體),逐一比對不可行——這是 3.2 節 Aho-Corasick 的主場:一次掃描、O(n + z)。
  2. 規避(adversarial):使用者用諧音(「賭博」→「賭愽」)、夾符號(「毒.品」)、全半形混用(123)、拆字(「木義」→「楊」類)繞過。工程對策:規格化前處理(全半形統一、去零寬字元與標點)、拼音/字形正規化(把字映射到拼音鍵,讓「愽」與「博」同鍵)、白名單(避免誤傷正常詞)。
  3. 誤判成本不對稱:漏放(false negative)傷平台、誤殺(false positive)傷使用者。因此系統常分級:黑名單(AC 自動機硬擋)→ 模糊規則(編輯距離容錯,4.3 節)→ 模型打分(9.1 節分類器或 LLM 審核)逐層遞進,僅可疑的才進入昂貴的下層。

另一個關鍵工程議題是詞庫熱更新:線上服務不能停機重建索引。AC 自動機的詞庫更新可採「雙緩衝」——後台建新樹、原子切換指標;若更新頻繁且詞庫超大,可改用 AC 自動機的增量變體或改用 DFA 併合。替換/遮蔽(顯示成 ***)只需記錄匹配位置,依「最長匹配優先」策略遮蔽。

本節實作:規格化管線 → AC 自動機過濾 → 拼音容錯層 → 遮蔽輸出。這個組合是產業敏感詞系統的骨架。

演算法步驟#

  1. 規格化:全形→半形、大寫→小寫、去除零寬字元(\u200b 系列)與干擾符(可配置白名單保留必要標點)。
  2. 建索引:敏感詞(含變體展開後的規格化形式)插入 AC 自動機,BFS 建失敗連結。
  3. 掃描:對規格化後文字跑 AC,收集所有匹配 (位置, 詞);同位置多匹配時取最長(「毒品交易」優先於「毒」)。
  4. 遮蔽:合併重疊區間後逐段替換為 *。
  5. 容錯層(可選):對未命中但含可疑子序列的文字,抽取字元的拼音首碼串,再跑一個「拼音版 AC 自動機」,命中則送人工/模型複審而非直接擋。

Python 程式碼範例#

from collections import deque
import re

class AC:
    def __init__(self):
        self.next = [{}]; self.fail = [0]; self.out = [[]]

    def add(self, word, tag=None):
        s = 0
        for ch in word:
            if ch not in self.next[s]:
                self.next.append({}); self.fail.append(0); self.out.append([])
                self.next[s][ch] = len(self.next) - 1
            s = self.next[s][ch]
        self.out[s].append(tag or word)

    def build(self):
        q = deque()
        for ch, v in self.next[0].items():
            q.append(v)
        while q:
            u = q.popleft()
            for ch, v in self.next[u].items():
                f = self.fail[u]
                while f and ch not in self.next[f]:
                    f = self.fail[f]
                self.fail[v] = self.next[f].get(ch, 0) \
                    if self.next[f].get(ch) != v else 0
                self.out[v] += self.out[self.fail[v]]
                q.append(v)

    def find(self, text):
        s, hits = 0, []
        for i, ch in enumerate(text):
            while s and ch not in self.next[s]:
                s = self.fail[s]
            s = self.next[s].get(ch, 0)
            for w in self.out[s]:
                hits.append((i - len(w) + 1, i + 1, w))
        return hits

# ---- 規格化:全形轉半形、去零寬與干擾符 ----
ZERO = dict.fromkeys(map(ord, "\u200b\u200c\u200d\ufeff\u00ad"), None)
def normalize(t):
    t = t.translate(ZERO)
    t = "".join(chr(ord(c) - 0xFEE0) if 0xFF01 <= ord(c) <= 0xFF5E else c
                for c in t)
    return re.sub(r"[\s.·*]+", "", t)

# ---- 拼音首碼鍵(極簡字典;實務用 pypinyin)----
PY = {"賭": "d", "博": "b", "博": "b", "愽": "b", "毒": "d", "品": "p"}
def pinyin_key(t):
    return "".join(PY.get(c, c) for c in t)

blocklist = ["賭博", "毒品", "詐騙"]
ac = AC()
for w in blocklist:
    ac.add(normalize(w))
    ac.add(pinyin_key(normalize(w)), tag="pinyin:" + w)   # 容錯層
ac.build()

def filter_text(text):
    norm = normalize(text)
    hits = ac.find(norm) + ac.find(pinyin_key(norm))   # 原文+拼音容錯層
    real = [h for h in hits if not h[2].startswith("pinyin:")]
    fuzzy = [h for h in hits if h[2].startswith("pinyin:")]
    # 最長匹配遮蔽
    mask = list(norm)
    for lo, hi, w in sorted(real, key=lambda h: -(h[1] - h[0])):
        for i in range(lo, hi):
            mask[i] = "*"
    verdict = "阻擋" if real else ("可疑→複審" if fuzzy else "通過")
    return verdict, "".join(mask)

for t in ["一起去賭博嗎", "毒.品交易", "賭愽一晚", "我去圖書館看書",
          "全形:賭博!", "線上詐騙很多"]:
    v, m = filter_text(t)
    print(f"{t} -> {v}  規格化遮蔽: {m}")

複雜度分析#

  • 掃描:O(n + z)(n 為規格化後長度、z 為匹配數)——5 萬詞庫下每百萬字毫秒級,這是選 AC 而非逐一比對的原因。
  • 建索引:O(總詞長);熱更新用雙緩衝 O(總詞長) 後台完成。
  • 規格化:O(n);拼音層讓字元集變大(拼音鍵),DFA 空間略增但掃描仍線性。
  • 誤判率管理:拼音容錯層召回↑、精確率↓,故只送複審不直接擋——分層讓「誤判成本」可控。

練習#

  1. 為系統加上白名單:建第二棵 AC 樹,若匹配區間完全落在白名單詞內則忽略(如詞庫有「解放」而白名單有「解放南路」),驗證遮蔽輸出正確。
  2. 實作重疊區間合併:把 find 的輸出按起點排序後合併重疊/相鄰區間,再一次性生成遮蔽字串,修正「多個匹配交疊時遮不乾淨」的 bug。
  3. 壓力測試:生成 10 萬字隨機文本(含 0.1% 真實敏感詞),量測規格化+AC 掃描的每秒吞吐量;再把詞庫從 1 千擴到 5 萬條,驗證掃描時間幾乎不變——這正是「索引與查詢解耦」的實證。