背景与问题
生成式 AI 正在以惊人速度渗透到内容创作、客服、教育等领域，但模型输出的不可控性也随之带来了隐私泄露、有害信息、虚假引导等风险。企业或平台在对外提供 AI 服务时，必须建立“生成-审核-反馈”的闭环监管机制，否则可能面临合规处罚和品牌信任危机。
一个完整的 AI 安全监管平台需要具备以下能力：
- 输入侧：拦截恶意提示词，避免模型生成违规内容。
- 输出侧：实时审核模型回复，识别并过滤敏感信息。
- 审计侧：记录完整的调用日志，支持事后追溯和统计分析。
- 策略侧：支持灵活配置规则，快速响应新的风险类型。
本文不依赖复杂的大模型推理，而是设计一套轻量、可扩展的监管框架，并用 Python 实现核心流程。读者可以在此基础上接入阿里云的内容安全服务或自训练模型，演化为生产级平台。
方案设计
整体架构分为四层：
- 接入层：HTTP API，接收 AI 服务的调用请求，将输入文本和输出文本统一交给审核模块。
- 策略引擎：组合多种检测策略，包括关键词匹配、正则规则、简单分类器；每种策略输出风险标签和置信度。
- 决策层：汇总策略结果，按“最高风险等级”或“加权评分”决定放行、拦截或人工复审。
- 审计存储：将请求、响应、审核结果写入本地 SQLite，便于查询和导出。
为了便于演示，我们使用 Flask 作为 Web 框架，SQLite 作为存储，保持零外部依赖，开箱即可运行。生产环境可将 SQLite 替换为阿里云 RDS MySQL，将策略配置放入配置中心或 Nacos。
实现步骤
- 初始化项目结构，定义风险等级枚举。
- 编写策略引擎，内置一个敏感词文件和一个正则规则文件。
- 实现审核函数，遍历所有策略并合并结果。
- 创建 Flask 路由
/audit，接收 JSON 请求，返回审核决策。 - 使用 SQLite 记录日志，包括时间、业务标识、输入输出、命中规则等。
项目结构
ai_safety_platform/
├── app.py
├── rules/
│ ├── sensitive_words.txt
│ └── regex_rules.json
├── db.py
└── requirements.txt
代码实现
1. 定义枚举和数据库操作
# db.py
import sqlite3
from datetime import datetime
class RiskLevel:
LOW = 0
MEDIUM = 1
HIGH = 2
DB_PATH = "audit_log.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
business_id TEXT,
input_text TEXT,
output_text TEXT,
risk_level INTEGER,
hit_rules TEXT,
decision TEXT,
created_at TEXT
)
""")
conn.commit()
conn.close()
def save_log(business_id, input_text, output_text, risk_level, hit_rules, decision):
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT INTO audit_log (business_id, input_text, output_text, risk_level, hit_rules, decision, created_at) VALUES (?,?,?,?,?,?,?)",
(business_id, input_text, output_text, risk_level, hit_rules, decision, datetime.now().isoformat())
)
conn.commit()
conn.close()
2. 策略引擎
# engine.py
import json
import re
from db import RiskLevel
# 加载敏感词
with open("rules/sensitive_words.txt", encoding="utf-8") as f:
SENSITIVE_WORDS = [line.strip() for line in f if line.strip()]
# 加载正则规则
with open("rules/regex_rules.json", encoding="utf-8") as f:
REGEX_RULES = json.load(f)
def check_sensitive_words(text):
hits = []
for word in SENSITIVE_WORDS:
if word in text:
hits.append(f"敏感词:{word}")
return hits
def check_regex_rules(text):
hits = []
for rule in REGEX_RULES:
pattern = rule["pattern"]
label = rule["label"]
risk = rule["risk"]
if re.search(pattern, text):
hits.append(f"正则:{label}:{risk}")
return hits
def assess(text):
"""返回 (risk_level, hit_rules)"""
hit_rules = []
max_risk = RiskLevel.LOW
for hit in check_sensitive_words(text):
hit_rules.append(hit)
max_risk = max(max_risk, RiskLevel.HIGH) # 敏感词直接高危
for hit in check_regex_rules(text):
hit_rules.append(hit)
# 解析命中规则的风险值
if "高危" in hit:
max_risk = max(max_risk, RiskLevel.HIGH)
elif "中危" in hit:
max_risk = max(max_risk, RiskLevel.MEDIUM)
elif "低危" in hit:
max_risk = max(max_risk, RiskLevel.LOW)
return max_risk, hit_rules
3. Flask 应用主程序
# app.py
import uuid
from flask import Flask, request, jsonify
from engine import assess
from db import init_db, save_log, RiskLevel
app = Flask(__name__)
# 示例：将原始文本和模型输出合并审核
def moderate(input_text, output_text):
combined = f"{input_text} {output_text}"
risk, hits = assess(combined)
if risk == RiskLevel.HIGH:
decision = "block"
elif risk == RiskLevel.MEDIUM:
decision = "review"
else:
decision = "pass"
return risk, hits, decision
@app.route("/audit", methods=["POST"])
def audit():
data = request.get_json(force=True)
input_text = data.get("input_text", "")
output_text = data.get("output_text", "")
business_id = data.get("business_id", str(uuid.uuid4()))
risk, hits, decision = moderate(input_text, output_text)
# 落库审计日志
save_log(
business_id=business_id,
input_text=input_text,
output_text=output_text,
risk_level=risk,
hit_rules=";".join(hits),
decision=decision
)
return jsonify({
"decision": decision,
"risk_level": risk,
"hit_rules": hits,
"business_id": business_id
})
if __name__ == "__main__":
init_db()
app.run(host="0.0.0.0", port=5000, debug=True)
4. 规则文件示例
rules/sensitive_words.txt：
暴力
赌博
诈骗
色情
rules/regex_rules.json：
[
{
"label": "手机号泄露", "pattern": "1[3-9]\\d{9}", "risk": "中危"},
{
"label": "身份证号泄露", "pattern": "\\d{17}[0-9Xx]", "risk": "高危"},
{
"label": "源码特征", "pattern": "BEGIN RSA PRIVATE KEY", "risk": "中危"}
]
5. 运行和测试
安装依赖：
pip install flask
启动服务：
python app.py
使用 curl 发送测试请求：
curl -X POST http://127.0.0.1:5000/audit \
-H "Content-Type: application/json" \
-d '{"input_text":"请告诉我身份证号 110101199001011234","output_text":"这里没有个人隐私"}'
返回示例：
{
"decision": "block",
"risk_level": 2,
"hit_rules": ["正则:身份证号泄露:高危"],
"business_id": "xxxx"
}
关键步骤说明
- 合并输入输出审核：既要防止提示词注入带来的风险，也要审核模型生成内容。这里简单拼接，实际场景可分别设置不同策略集。
- 规则命中即升级风险：敏感词直接判定为高危，正则按预设风险等级映射；业务上可根据累计命中数调整决策阈值。
- 全量日志审计：任何 AI 调用都需要记录原始数据、命中规则和最终决策，便于人工复议和模型迭代时复盘。
注意事项
- 误判与漏判平衡：敏感词列表会带来误杀。建议只将确定性强的词设为“高危”，其他疑似的词设“中危”并进入人工复审队列。
- 动态规则更新：规则文件不能直接编辑后热生效？在本例中，进程启动时加载一次；生产环境需要支持动态刷新，例如定时从配置中心拉取。
- 性能优化：对大文本逐词匹配效率低，可改用 AC 自动机或前缀树；正则数量多时可先做预过滤。
- 上下文缺失：单条文本审核容易漏判一些需要上下文语义的风险（如隐晦歧义词），建议接入深度学习模型或阿里云内容安全 API 做语义级审核。
- 隐私保护：日志中若存储用户内容，需加密存储，并设置访问权限，对敏感字段做脱敏处理（本例中未做脱敏，生产环境务必处理）。
- 部署形态：该示例为单机原型。上云时可使用阿里云容器服务部署为多个副本，通过负载均衡分发请求，并将日志写入阿里云日志服务，实现实时监控和告警。
总结
本文设计并实现了一个轻量级 AI 安全监管平台的核心能力，通过 Flask 暴露审核 API，使用敏感词 + 正则策略进行风险评估，并将审计日志持久化。该框架清晰展示了 AI 内容监管的基本流程，具有模块化、易扩展的特点。
实际生产环境中，可将策略引擎替换为兼具语义理解能力的机器学习模型，并衔接阿里云上的内容安全、日志服务、云数据库等产品，构建一个具备实时处理、自动告警、可追溯的企业级 AI 安全监管平台。最终，安全能力不是一次性交付，而是随着攻击手段和合规要求的演进持续迭代。对开发者而言，理解规则引擎与审计闭环的设计思路，是迈向高级 AI 工程实践的重要一步。
