敏感内容识别

审核不只是”鉴黄”

鉴黄检测上线后，我以为审核问题解决了。直到有一天，一个用户上传了一张看似正常的风景照——画面上是一条安静的街道，但街道旁的广告牌上写着一句违禁标语。

AI 鉴黄模型判定：安全。

因为鉴黄模型只看图像的视觉内容，完全看不到图片里的文字。

我又翻了翻后台数据，发现了更多被鉴黄模型放过的”漏网之鱼”：

已通过鉴黄但可能违规的图片：
1. 风景照 + 广告牌上的违禁标语     → 文字违规
2. 暴力场景截图（电影/新闻截图）    → 暴力血腥
3. 政治人物合影                    → 政治敏感
4. 假冒伪劣商品照片                → 广告违规
5. 含有二维码的图片                → 引流/诈骗

鉴黄只解决了色情这一个维度。一个完整的内容审核系统，需要多维度、多模态的检测能力。

维度一：OCR 文字识别

图片中的文字是审核的重要维度。很多违规内容不以视觉形式呈现，而是以文字形式”藏”在图片里。

OCR 方案对比

对比测试：

OCR 方案对比（100 张含文字的图片）

方案              识别准确率   速度      成本
──────────────────────────────────────────────
Tesseract         72%        800ms     免费
阿里云 OCR         96%        200ms     0.8元/千次
腾讯云 OCR         95%        250ms     0.7元/千次

结论：Tesseract 对中文的识别准确率太低（72%），特别是手写体、艺术字体、小字。云 API 是更好的选择。

文字内容审核

识别出文字后，还需要判断文字内容是否违规：

OCR 识别结果
    │
    ▼
归一化：去空格、全角转半角、去零宽字符、大小写统一
    │
    ▼
三层判定（从便宜到贵）
    ├─ 第一层：正则 / 关键词库     —— 微秒级，覆盖绝大多数明确违规
    ├─ 第二层：语义模型（云 API）  —— 百毫秒级，处理变体和隐喻
    └─ 第三层：人审               —— 模型的灰色地带

OCR 识别出的文字不能直接送进审核接口——广告牌上的字经常被拆成一行一行的碎片，或者夹杂空格和符号，直接送过去大概率判不出来。所以第一件事是归一化：

import re
import unicodedata


def normalize_text(text: str) -> str:
    """全角转半角、去零宽字符、空白折叠、英文转小写。"""
    text = unicodedata.normalize("NFKC", text)
    text = text.replace("\u200b", "").replace("\ufeff", "")
    text = re.sub(r"\s+", " ", text)
    return text.strip().lower()

归一化之后，第一层用关键词库做快速拦截。这层几乎不花钱，而且能挡掉绝大多数一眼就能认出来的违规内容：

# 禁用词库分级：一级直接拦截，二级转人审
BLOCK_WORDS = {
    "赌博", "博彩", "贷款", "办证", "发票", "代开发票",
}
REVIEW_WORDS = {
    "医疗", "保健品", "金融", "兼职", "招聘",
}


def keyword_check(text: str) -> tuple[str, list[str]]:
    """返回 (动作, 命中的词)。一级词直接拦截，二级词转人审。"""
    hits = [word for word in BLOCK_WORDS if word in text]
    if hits:
        return "reject", hits
    hits = [word for word in REVIEW_WORDS if word in text]
    if hits:
        return "review", hits
    return "pass", []

第二层用云端的文本审核接口。这一层的价值在于处理”没有明显关键词”但语义违规的情况——比如”一步到位，无需抵押”这种变体。阿里云内容安全的文本审核接口是 TextModeration：

import json

from aliyunsdkcore.client import AcsClient
from aliyunsdkgreen.request.v20180502.TextModerationRequest import TextModerationRequest


def moderate_text(client: AcsClient, content: str) -> dict:
    """调用阿里云内容安全文本审核，返回处置建议和命中的标签。"""
    request = TextModerationRequest()
    request.set_accept_format("json")
    request.set_content_type("application/json")
    request.set_Method("POST")
    request.set_HttpProtocol("HTTPS")

    payload = {
        "content": content,
        "scene": "ugc",                 # 社交 UGC 场景
        "serviceParameters": {
            "keywordDetection": True,   # 关键词
            "languageDetection": True, # 语言
            "textModeration": True,    # 语义
        },
    }
    request.set_body(json.dumps(payload))

    response = json.loads(client.do_action_with_exception(request))
    data = response.get("data", {})
    labels = response.get("labels", [])
    return {
        # 接口返回的建议动作：pass / review / block
        "suggestion": data.get("suggestion", "pass"),
        "labels": [
            {"label": item.get("label"), "probability": item.get("probability")}
            for item in labels
        ],
    }

三层判定的结果要合成一个最终动作，规则和图片审核那一节是同一套思路——机器不直接决定拦截，转人工兜底：

最后补一条工程上的要求：文字审核的判定依据同样要写进审核记录。命中的关键词、模型的标签和概率、当时生效的词库版本——三样缺一不可。否则半年后用户来申诉，你根本没法解释当时为什么判了违规。

还有一点：OCR 识别出来的文字里，正常的拍摄参数（ISO 400、f/2.8、2024-06-01）非常常见。这些不是违规，但会稀释语义模型的注意力。所以送审前要先做一轮”拍摄参数过滤”，把这类明显无害的内容剔除掉。

维度二：暴力血腥检测

暴力检测的难点：

容易误判的场景：
- 烹饪节目（生肉、鲜血）      → 误判为暴力
- 医学手术照片               → 误判为血腥
- 体育比赛（拳击、摔跤）     → 误判为暴力
- 万圣节装扮                 → 误判为恐怖
- 新闻报道中的事故现场       → 误判为血腥

这些场景的处理策略：
- 医学/新闻类：加标签但不拦截
- 体育比赛：加"体育"标签后降低权重
- 用户标注的"恐怖电影截图"：允许但加内容警告

维度三：政治敏感检测

这是最敏感也最必须的维度。国内的平台必须具备政治敏感内容的识别能力。

维度四：广告和二维码检测

摄影社区里经常有人上传带二维码的图片做引流：

上传的图片
    │
    ▼
扫描二维码
    ├─ 扫到 → 解码出内容
    │         ├─ URL      → 展开短链 → 检查域名
    │         ├─ 纯文本   → 送文字审核
    │         └─ 联系方式 → 按引流规则处理
    └─ 没扫到 → 是否含明显广告元素？（文字、促销模板）

二维码检测本身不难，麻烦的是扫到之后怎么办。扫出内容只是第一步，真正要判断的是”这个内容该不该出现在摄影社区里”。

先看检测。OpenCV 自带的二维码检测器不需要额外依赖：

import cv2
import numpy as np


def detect_qr_codes(image_bytes: bytes) -> list[str]:
    """返回图片中所有能解码出的二维码内容，解不出来就返回空列表。"""
    buffer = np.frombuffer(image_bytes, dtype=np.uint8)
    image = cv2.imdecode(buffer, cv2.IMREAD_COLOR)
    if image is None:
        return []

    detector = cv2.QRCodeDetector()
    found, points, _ = detector.detectAndDecodeMulti(image)
    if found:
        return [text for text in found if text]

    # 多码检测失败时，退回单码检测
    text, points, _ = detector.detectAndDecode(image)
    return [text] if text else []

解码出来的内容要分类处理：

短链展开要小心：不能自动访问未知域名。展开动作本身就是一个 SSRF 风险源，展开请求要发往一个隔离的探测服务，带超时和出口 IP 限制，并且只跟随有限次跳转。

域名判断靠一张表，简单但有效：

from urllib.parse import urlparse

# 允许出现的域名：用户自己的站点、已认证的机构主页
ALLOWED_HOSTS = {
    "guangying.example.com",
    "blog.example.com",
}

# 明确禁止的域名：博彩、诈骗、灰产
BLOCKED_HOSTS = {
    "example-betting.com",
    "spam-domain.net",
}

# 高风险但不直接拦截的关键词，命中转人审
SUSPICIOUS_KEYWORDS = ("bit.ly", "qr", "pay", "vip", "gift")


def judge_url(url: str) -> str:
    """返回 reject / review / pass。"""
    host = (urlparse(url).hostname or "").lower()
    if host in BLOCKED_HOSTS:
        return "reject"
    if host in ALLOWED_HOSTS:
        return "pass"
    if any(word in host for word in SUSPICIOUS_KEYWORDS):
        return "review"
    return "review"

广告检测是另一个维度。除了二维码，还有大量不带二维码的广告：白底商品图加一行促销文字、统一的促销模板、明显的联系方式水印。这些靠视觉模型判断比较准，但成本也最高。

我把它放在最后一层，作为兜底：

最后那行的 75% 意味着不能拿它直接做拦截决策。所以视觉广告检测的命中结果只做两件事：给图片加”可能含广告”的标签，以及降低它在推荐流里的权重。真正要拦截的，还是靠上面几层那些高置信度的信号。

多维度审核整合

把所有审核维度整合到一个服务中：

                     ┌────────────────────────────┐
   图片字节流  ──────▶  审核编排服务               │
                     │                            │
                     │  1. 并行派发各维度         │
                     │  2. 每个维度独立超时       │
                     │  3. 汇总结果并合并动作      │
                     │  4. 写审核记录             │
                     └────────────┬───────────────┘
                                  │
        ┌────────────┬────────────┼────────────┬────────────┐
        ▼            ▼            ▼            ▼            ▼
   鉴黄模型      OCR + 文字     暴力检测    二维码/广告   视觉广告
   (150ms)       (250ms)        (200ms)      (80ms)       (600ms)
        │            │            │            │            │
        └────────────┴────────────┴────────────┴────────────┘
                                  │
                     合并规则：取最严格的动作
                     耗时：取最慢的那个维度

整合的第一个要点是并行。五个维度如果串行做，耗时是所有维度之和；并行做，耗时是最慢的那个维度。图片审核对延迟敏感——用户上传后要等结果，这段时间直接影响发布体验。

第二个要点是每个维度独立超时。某个维度卡住，不能拖死整条链路。超时的维度按”降级策略”处理，而不是让整次审核失败。

import asyncio
from dataclasses import dataclass
from typing import Awaitable, Callable

# 各维度的超时时间和降级策略
# 降级策略：fail_closed = 出问题时转人审，fail_open = 出问题时放行
DIMENSIONS: dict[str, dict] = {
    "porn":     {"timeout": 1.0, "on_timeout": "fail_closed"},
    "ocr":      {"timeout": 2.0, "on_timeout": "fail_closed"},
    "violence": {"timeout": 2.0, "on_timeout": "fail_closed"},
    "qrcode":   {"timeout": 0.5, "on_timeout": "fail_open"},
    "ad":       {"timeout": 3.0, "on_timeout": "fail_open"},
}


@dataclass
class DimensionResult:
    name: str
    action: str          # pass / review / reject
    score: float
    reason: str
    timed_out: bool = False


async def run_dimension(
    name: str,
    check: Callable[[], Awaitable[DimensionResult]],
) -> DimensionResult:
    """跑单个维度，超时按配置降级，绝不把异常抛给编排层。"""
    config = DIMENSIONS[name]
    try:
        return await asyncio.wait_for(check(), timeout=config["timeout"])
    except asyncio.TimeoutError:
        action = "review" if config["on_timeout"] == "fail_closed" else "pass"
        return DimensionResult(name, action, 0.0, "timeout", timed_out=True)
    except Exception as exc:
        action = "review" if config["on_timeout"] == "fail_closed" else "pass"
        return DimensionResult(name, action, 0.0, f"error: {exc}", timed_out=True)


async def moderate(image_bytes: bytes, checks: dict) -> list[DimensionResult]:
    """并行跑所有维度。"""
    tasks = [run_dimension(name, check) for name, check in checks.items()]
    return list(await asyncio.gather(*tasks))

降级策略不能一刀切。高危维度必须 fail_closed——鉴黄、暴恐这类维度挂了就转人审，不能放行；低危维度可以 fail_open——二维码检测挂了先放行，否则一次服务抖动就会让整个平台的图片审核停摆。

第三个要点是结果合并。规则很简单：取最严格的动作。任何一个维度判了 reject，整张图就是 reject；没有 reject 但有 review，就是 review。

ACTION_PRIORITY = {"pass": 0, "review": 1, "reject": 2}


def merge(results: list[DimensionResult]) -> tuple[str, list[DimensionResult]]:
    """合并各维度结果：动作取最严格，命中的原因一并返回。"""
    action = max(results, key=lambda r: ACTION_PRIORITY[r.action]).action
    hits = [r for r in results if r.action != "pass"]
    return action, hits

三个工程上必须做的事：

1. 缓存重复审核。同一张图片可能被反复上传（用户重传、客户端重试）。用图片内容的哈希做键缓存审核结果，命中缓存就跳过所有模型调用，既省钱也降延迟。缓存要带 TTL，模型版本或阈值变更时按版本号失效。
2. 审核记录要完整。每个维度的动作、分数、原因、超时与否，全部写进审核记录。前面说的”分数、阈值、模型版本”三件套在这里扩展成”每个维度一份”。出问题时能立刻回答”是哪一维判的”。
3. 维度是可以增删的。新增一个维度不应该改编排层的代码——维度的超时和降级策略放在上面那张配置表里，加一行就是加一个维度。编排逻辑永远只认”并行执行加取最严”这条规则。

性能优化

多维度审核意味着多次 API 调用，如何控制延迟？

进一步的优化——分级审核：

上一章节

鉴黄检测

下一章节

审核流程设计