AI生成内容合规过滤系统设计与实现：敏感词、规则引擎与审计日志全解析

前阵子我所在的团队刚刚完成一个面向客户的内容审核系统升级，主要是把生成式AI服务接入到正式业务流程之前，需要解决一个很实际的合规问题：怎么在产品侧把内容安全这件事做扎实。花了两周时间把规则引擎、敏感词库、图像审核、审计日志这几块全部重做了一遍，踩了不少坑，也沉淀下来一套可以直接复用的方案。今天就把这套“AI生成内容合规过滤系统”从设计到落地的完整思路分享出来。

1. 内容整体设计与思路拆解

先说说这套系统是干什么的。简单来讲，当你的产品里接入了AI生成能力——比如智能客服自动回复、营销文案生成、社区帖子标题推荐——你不可能让模型输出的每一句话都直接展示给用户，必须经过一道内容安全的过滤闸门。这道闸门要做的事情分三层：

• 第一层，拦截明显违规的词汇和短语，比如暴力、色情、歧视等敏感表达；
• 第二层，对语义层面进行判断，比如“这家店菜量小得离谱”和“这家店该倒闭”是两种完全不同的情绪，需要区分；
• 第三层，留下完整的审计记录，万一出现纠纷，可以追溯某条内容是什么时候、由哪个模型、基于什么输入生成的。

这个思路对应的就是一个典型的“规则+模型+审计”三层架构。实际落地时，绝大多数团队会把第一层做得很重，因为规则引擎可控性最强、响应最快、解释成本最低。第二层需要引入辅助分类模型，成本和时延都会上来，所以通常作为补充而不是主力。第三层则是很多团队容易忽略的，我在这里先提醒一句：审计日志不是可有可无的，后面做问题回溯的时候你就知道它的价值了。

核心需求拆解下来其实就是几句话：第一，哪些词必须一刀切拦截；第二，哪些表达需要模型判断；第三，每次审核的结果怎么存、怎么查、怎么统计。把这三件事想清楚，整个系统的骨架就不会歪。

方案选型上，我们最终选择了“本地规则引擎+云端分类API兜底”的组合。原因很现实：纯本地规则覆盖不了语义场景，纯云端API又会有网络延迟和单次调用成本的问题，而且对方一旦接口调整，你这边就是线上事故。两者结合，规则引擎挡住绝大多数流量，只有拿不准的内容才交给分类API，既省钱又快。

2. 核心细节解析与实操要点

2.1 敏感词库的分级策略

敏感词库是整个系统的基础，但很多人一上来就犯一个错误：把所有词都放在一个桶里，不分等级。后果就是，有些其实是正常语境下的词也被误杀，用户体验极差。我建议把词库拆成三个级别：

这个分级的意义在于，把“必定拦截”和“需要思考”分开。一级词库尽量做得小而准，宁可漏掉一些模糊表达，也不要错误拦截正常句子。二级词库可以做得大一些，反正是交给模型去判断，不会一刀切。三级词库是新词发现的主要来源，很多擦边词会先在这里跑一段时间，靠人工确认升到二级或一级。

词库的维护频率也很重要。我见过不少团队把词库建完就扔在那里半年不更新，这基本等于让系统裸奔。正确的做法是每周至少过一次“新增词候选”，来源包括：用户举报、审核后台的标记数据、外部公开的违规案例。词的变体也要处理，比如同音字替换、拼音替换、谐音梗，这些都需要通过规则引擎里的“变体匹配”能力覆盖。

2.2 规则引擎的匹配策略

规则引擎听起来高大上，其实核心就是字符串匹配加上一些正则规则。但真正的复杂度在“怎么匹配”和“匹配到什么程度”。

有个词叫“部分匹配”。举个例子，“假货”是一级词，但“这包是假货”和“假货贩子都该抓”显然是两种语境。人眼可以判断，规则引擎不行。所以我做了两套匹配模式：精确匹配和扩展匹配。精确匹配要求整词命中，扩展匹配则允许词根命中后进入二级判断。这样做的好处是可以大幅减少误杀。

再一个关键是“上下文窗口”。我这边实现的时候，不只是匹配词本身，还会把该词前后各N个字符作为上下文一起送入语义模型。N的取值建议在10到20之间，太短上下文信息不足，太长又引入过多噪声。这个参数一开始就配好，后面在调优阶段会省很多时间。

规则引擎的性能也要注意。它的输入是每一条待审核的内容，如果每次都做几十万词库的全量匹配，性能一定扛不住。我们的实践是给词库建索引，按词的首个字做哈希分桶，命中一个桶之后再做精确匹配。这个优化做完，单条内容的匹配耗时从几十毫秒降到了个位数毫秒。

2.3 语义分类模型的选择

规则引擎挡完之后，剩下的内容需要一个分类模型来兜底。这里讲的不是训练一个大模型，而是用已有的成熟模型做推理分类。我们的场景只需要判断“是否违规”，所以用二分类模型就够了。但如果你需要细分违规类型，比如区分是色情还是暴力，那就得用多分类模型。

模型选型的时候我对比过几类方案：

对于多数中小团队，我的建议是先上云端API，等业务量上来了再评估自训练模型。这不是技术站队的问题，而是资源分配的现实判断。标注数据这件事的隐性成本很容易被低估，一个分类模型没有几千条高质量标注样本，效果基本不忍直视。

分类结果的处理上，我建议不要直接以“概率值”做唯一判断依据。可以设一个双阈值：高于高阈值直接拦截，低于低阈值直接放行，中间地带进入人工审核队列。这样一来，模型的不确定性就被显式地交给人工兜底，而不是让AI替你做决定。

2.4 审计日志的设计

审计日志这一个模块，我愿称它为整条链路里性价比最高的部分。原因很简单，大量的规则和模型误判问题，最后都是靠日志回溯才发现根因的。没有日志，你只能听到用户反馈“内容发不出去”，但完全不知道为什么发不出去。

日志需要记录的信息包括：待审核内容的完整原文、审核结果、命中的规则或模型名称、规则命中的具体词、模型的置信度、请求来源、请求时间。这些字段缺一个，都可能在追溯时遇到盲区。

存储方面，日志表的写入量会比较大，建议按天分表，同时设置定期归档。查询的需求通常是“按时间范围+按内容关键字”，所以索引要建在这两个字段上。UI层面，给运营同学配一个简单的搜索页面，让她们能按时间、按结果、按来源筛选，减少对技术团队的依赖。

3. 实操过程与核心环节实现

3.1 环境准备

本方案基于Python实现，原因有二：一是团队技术栈以Python为主，二是后续如果要接入机器学习模型，生态更顺。需要的依赖如下：

• Python 3.10+
• flask（轻量API服务框架）
• sqlite3（日志存储，量级上来后可换MySQL）
• 正则表达式模块re（标准库，无需安装）

依赖非常简单，跑起来不需要重型框架。如果你所在团队用的是Java或Go，对应换成Spring Boot或Gin即可，架构思路完全一致，不需要照搬代码。

3.2 规则引擎核心代码

规则引擎我们分两步实现，先是敏感词管理模块。

from typing import Dict, List, Tuple class SensitiveWordManager: """敏感词库管理器，按级别存储并完成构建索引""" def __init__(self): # 分三个级别存储词库，每个级别的词是一个list self.word_dict: Dict[int, List[str]] = { 1: [], # 一级：直接拦截 2: [], # 二级：进入语义模型判断 3: [], # 三级：记录日志但放行 } # 索引：首字 -> 该首字对应的全部词 self.index: Dict[str, List[Tuple[str, int]]] = {} def add_word(self, word: str, level: int) -> None: """新增词：存入对应级别，同时建立首字索引""" self.word_dict[level].append(word) first_char = word[0] if first_char not in self.index: self.index[first_char] = [] self.index[first_char].append((word, level)) def load_from_file(self, filepath: str, level: int) -> None: """从文件批量加载词库，每行一个词，支持#注释""" with open(filepath, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): self.add_word(line, level) def match(self, text: str) -> List[Tuple[str, int, int]]: """ 匹配文本中包含的敏感词，返回 [(词, 级别, 出现位置)] 先根据首字索引定位候选词，再做字符串包含验证 """ results = [] for i in range(len(text)): current_char = text[i] if current_char not in self.index: continue for word, level in self.index[current_char]: if text.startswith(word, i): # 如果已经记录过同一个词，跳过，避免重复添加 if any(w[0] == word for w in results): continue results.append((word, level, i)) return results

这段代码的逻辑并不复杂，核心是首字索引。举个例子，如果词库里有“假货”和“假一赔十”，那么在匹配“这批货假一赔十”时，从第一个字符开始遍历，遇到“假”字就只比对以“假”开头的候选词，而不用全量扫词库。这个优化在词库数量过万后效果非常明显。

紧接着是审核服务层。它的职责是串起整个审核流程：先做规则匹配，再根据级别决定走向。

import re import hashlib import time from typing import Dict, List class AuditService: """审核服务，负责把规则匹配、模型判断、日志存储串起来""" def __init__(self, word_manager: SensitiveWordManager, model_client=None): self.manager = word_manager self.model_client = model_client # 语义模型API客户端，可选 self.audit_log = [] # 内存日志，便于演示 def _shadow_copy(self, text: str, word: str, position: int) -> str: """构造脱敏后的文本：命中敏感词的字符替换为*""" return text[:position] + "*" * len(word) + text[position + len(word):] def audit(self, content: str, source: str = "unknown") -> Dict: """ 单条内容审核主流程 返回：审核结果、命中详情、脱敏文本、耗时等信息 """ start = time.time() matched = self.manager.match(content) # 1. 统计各级命中数量 level1_hits = [w for w, lv, pos in matched if lv == 1] level2_hits = [w for w, lv, pos in matched if lv == 2] # 2. 决策逻辑 if level1_hits: result = "block" reason = f"命中一级敏感词: {','.join(level1_hits)}" elif level2_hits: # 二级命中：调用语义模型判断 if self.model_client: model_verdict = self.model_client.judge(content, level2_hits) result = "block" if model_verdict["is_violate"] else "pass" reason = f"语义模型判定: {model_verdict['label']}, 置信度={model_verdict['confidence']}" else: # 没有模型客户端，走安全策略：宁拦勿放 result = "block" reason = f"命中二级敏感词且未启用模型: {','.join(level2_hits)}" else: result = "pass" reason = "无敏感词命中" # 3. 脱敏展示文本 display_text = content for word, lv, pos in matched: if lv <= 2: # 一级二级词都脱敏，三级词保留用于观察 display_text = self._shadow_copy(display_text, word, pos) # 4. 记录审计日志 log_entry = { "time": time.strftime("%Y-%m-%d %H:%M:%S"), "content": content, "display_text": display_text, "result": result, "reason": reason, "source": source, "cost_ms": int((time.time() - start) * 1000), } self.audit_log.append(log_entry) return { "result": result, "reason": reason, "display_text": display_text, "matched_words": [{"word": w, "level": lv} for w, lv, pos in matched], }

这里有一个设计细节值得强调：当二级敏感词命中但没有配置模型的时候，我故意选择了“宁拦勿放”。这个策略在业务初期尤其是刚上线时非常关键。原因很简单，拦错一个正常内容，用户顶多觉得系统笨；放错一个违规内容，事情就不只是产品体验层面的问题了。等模型接入并且经过一段时间的准确率验证后，再逐步把策略切回“模型判断”。

3.3 语义模型客户端封装

模型客户端我们做一个简单的HTTP调用封装。这里假设你用的是某个云端分类API，实际使用中替换成你自己的接口即可。

import requests from typing import Dict, List class SemanticModelClient: """语义分类模型客户端封装""" def __init__(self, api_url: str, api_key: str, high_threshold: float = 0.8, low_threshold: float = 0.4): self.api_url = api_url self.api_key = api_key self.high_threshold = high_threshold self.low_threshold = low_threshold # 低阈值用于人工审核决策 def judge(self, text: str, matched_words: List[str]) -> Dict: """ 调用模型判断文本是否违规 返回结果包含 label 和 confidence """ payload = { "text": text, "matched_words": matched_words, } headers = {"Authorization": f"Bearer {self.api_key}"} # 实际调用，失败时默认按违规处理，保证安全 try: resp = requests.post(self.api_url, json=payload, headers=headers, timeout=3) resp.raise_for_status() data = resp.json() return { "is_violate": data["label"] == "violate", "label": data["label"], "confidence": data["confidence"], } except Exception as e: # 模型不可用时的安全兜底策略 return { "is_violate": True, "label": "unknown", "confidence": 1.0, "error": str(e), }

模型判断的阈值设置要结合业务容忍度来定。高阈值0.8意味着模型只有80%以上把握判定违规时才直接拦截，有效减少误杀。低阈值0.4以下直接放行。两者之间进入人工审核队列。这个双阈值策略比单阈值更符合真实业务场景，因为模型输出的概率并不代表真实概率，尤其是在分布外样本上。

3.4 完整的审核流程演示

把上面的模块串起来，模拟一条完整审核链路。我用两个例子演示，一个是一级词命中，一个是二级词命中。

if __name__ == "__main__": # 初始化词库 mgr = SensitiveWordManager() # 一级词示例 mgr.add_word("暴力", level=1) mgr.add_word("违禁品", level=1) # 二级词示例 mgr.add_word("假货", level=2) mgr.add_word("骗子", level=2) # 初始化模型客户端（模拟） client = SemanticModelClient( api_url="https://your-api.example.com/classify", api_key="test-key", ) audit_service = AuditService(mgr, client) # 测试1：一级词直接拦截 r1 = audit_service.audit("这个商家卖违禁品，大家注意", source="comment") print("测试1结果:", r1["result"], "|", r1["reason"]) print("脱敏文本:", r1["display_text"]) print("-" * 50) # 测试2：二级词走模型判断（这里模型不可达，走安全兜底） r2 = audit_service.audit("有人说这家店的货是假货", source="comment") print("测试2结果:", r2["result"], "|", r2["reason"]) print("脱敏文本:", r2["display_text"]) print("-" * 50) # 测试3：正常内容直接放行 r3 = audit_service.audit("这家店的服务态度不错，推荐", source="comment") print("测试3结果:", r3["result"], "|", r3["reason"])

如果模型API不可达，二级词的兜底策略会让它变成拦截，这符合安全优先原则。但我也要提醒：这个兜底策略在生产环境要配合告警使用。如果模型API持续不可达，系统会大面积拦截正常内容，这本身就是一次需要马上处理的事故。所以模型客户端里必须有超时控制和失败计数，触发连续失败后立刻返回错误给上层，而不是默默兜底。

3.5 参数选择与实际调优

参数这块我专门列几张表，方便大家对照着做。第一张是规则引擎相关参数：

第二张是模型参数。模型参数的核心是阈值，这个没有普适的最佳值，必须基于你自己的数据调。我的建议是准备一个包含200~500条内容的测试集，人工标注好是否违规，然后用不同阈值跑一遍，看准确率和召回率曲线，找一个最合适的折中点。

调优时还有一个免不了的话题：误杀和漏判怎么平衡。我个人经验是，与其追求完美平衡，不如设定清晰的红线。比如“漏判率不能超过1%”是红线，“误杀率不超过5%”是目标。然后倒推阈值在哪。这个做法比拍脑袋定阈值靠谱得多。

4. 审核API服务的搭建

规则引擎和模型客户端写好了，还需要把它暴露成一个接口供业务侧调用。这里我用Flask写一个极简的HTTP API。

from flask import Flask, request, jsonify import threading app = Flask(__name__) # 全局审核服务实例 audit_service = None @app.route("/audit", methods=["POST"]) def audit(): data = request.get_json() content = data.get("content", "") source = data.get("source", "unknown") if not content: return jsonify({"error": "content为空"}), 400 result = audit_service.audit(content, source=source) return jsonify(result) @app.route("/health", methods=["GET"]) def health(): return jsonify({"status": "ok"}) def initialize(): global audit_service mgr = SensitiveWordManager() # 加载词库：实际生产环境中从这里读取配置文件或数据库 mgr.load_from_file("words_level1.txt", level=1) mgr.load_from_file("words_level2.txt", level=2) # 初始化模型客户端 client = SemanticModelClient( api_url="https://your-api.example.com/classify", api_key="test-key", ) audit_service = AuditService(mgr, client) if __name__ == "__main__": initialize() app.run(host="0.0.0.0", port=8000)

接口设计上就两个端点，一个是审核接口，一个是健康检查。审核接口接收两个参数，content是必填的待审核内容，source是来源标识。返回结果里有result、reason、display_text和matched_words。业务方拿到的就是这四个字段。

关于接口返回的数据格式，有一个点值得留意：display_text这个东西，是给需要“展示脱敏后内容”的场景用的。比如社区帖子被拦截后，如果想告诉用户“你的内容里个别词汇不符合要求”，可以先把脱敏后的版本展示出来，让用户自己看是哪个词出了问题。这个小设计能明显减少用户因“莫名其妙被删”而产生的困惑。

并发方面，当前这版直接用了Flask内置的开发服务器，只适合测试环境。生产环境至少要用Gunicorn跑多worker，同时注意审核服务里词库索引的内存复制问题。多进程部署时，每个worker都会加载一份词库，内存占用要提前评估。词库如果达到10万量级，一个worker大概占100~200MB内存，三个worker就是600MB左右，这个成本也要算进部署规划里。

5. 常见问题与排查技巧实录

5.1 误杀率过高，正常内容被频繁拦截

出现这个问题，先别急着动阈值。第一步应该去审计日志里拉数据，看看被拦截的内容里，是哪些词贡献了绝大部分误杀。很多时候你会发现，就是那么十来个词造成的。把它们从一级降到二级，甚至加入三级词库，误杀率立刻大幅下降。

如果是模型判断导致的误杀，那就需要看模型返回的置信度分布。如果拦截的内容confidence普遍在0.6~0.7之间，说明高阈值设得太低，提到0.9试试。如果提阈值之后漏判案例变多，那就不是阈值的问题了，可能是二级词库太宽泛，需要收缩二级词库的收词标准。

5.2 变体词绕过规则引擎

同音替换、拼音替换、谐音梗是绕规则引擎最常用的手段。比如“违禁品”改成“wei jin pin”，或者用数字“520”暗示“我爱你”这类表达。应对手段是规则引擎里增加“变体匹配器”，对文本做归一化后再匹配。

归一化包含几步：全角转半角、大写转小写、去除零宽字符、拼音序列化。做完这四步，很多变体就会被打回原形。代价是归一化需要额外耗时，所以只建议对二级词库做变体匹配，一级词库保持原样精确匹配。

5.3 模型API抖动导致审核延迟

模型API不可达或者响应慢，直接影响审核链路延迟，用户端体感就是“发一条内容要转好几秒”。我踩过一次坑，当时没有给模型调用设置超时，整个请求挂了十几秒，业务侧直接超时，用户一脸懵。

解决方案就是我在封装代码里写的那条timeout=3，以及失败降级策略。任何外部调用都必须有超时控制，这是通用原则。同时要加熔断：连续失败超过一定次数（比如10次），直接切断模型调用，走“拦死”策略，同时触发告警让人介入。这样即使模型挂掉，也只是误杀多一点，不至于拖垮整个服务。

5.4 高频热点词冲击审核性能

有个很有意思的坑，就是某些热点事件发生后，大量内容集中在某几个新词上。这些词不在词库里，规则引擎会全部放行，等于是裸奔状态。等我们发现并加入词库，热点都过去了。这个问题的本质是词库更新不够快。

解决思路分两步：一是给三级词库建立“高频未命中词”的统计，凡是在较短时间内出现次数超过阈值的词，自动提升到二级词库待人工确认。二是配置词库热更新接口，运营同学在管理后台加词后，不需要重启服务，词库索引在内存里热替换。这两步做好，热点事件的响应时间能从按天级缩小到按小时级。

5.5 常见问题速查表

上面这张表里的每一行，背后都是我实际踩过的坑。比如脱敏文本重叠覆盖这个问题，当时真是没料到：两个敏感词位置相邻，第一个词替换成星号之后，第二个词的位置判断就错位了，出现“**暴露”这种诡异结果。后来改成先排序所有命中位置，再统一替换才解决。

6. 一点实践体会

做了这套系统之后，我最大的感受是：内容安全的核心难点，不在于算法有多先进，而在于你对业务场景的理解有多深。同样的一个词，在社区评论里和金融产品说明里的风险等级完全不一样。规则引擎也好，语义模型也好，都是工具，真正决定系统好坏的是规则的设计是否贴合你的业务。

另外想提醒一点，这套系统的效果评估不能只盯着准确率。我在实际运营里更看重两个指标：一个用户申诉率，即被拦截用户中发起申诉的比例；另一个是漏判响应速度，即违规内容发布后到被处理的时间差。这两个指标直接反映真实体验，比离线测试集上的准确率有说服力得多。

最后说一个常被忽略的细节：所有拦截记录的保留期限。最好设定明确的保留周期，比如90天或180天，到期由定时任务清理归档。别把所有数据都永久堆着，既浪费存储，也可能让系统在查询时越来越慢。做一个称职的“守门员”，既要拦得准，也要记得住，还要跑得快。