审核流程设计

一次误判引发的”事故”

审核系统上线后的第三天,摄影师老张给我发了一封长长的邮件。

他上传了一组人体艺术摄影——黑白光影、专业布光、构图讲究。AI 审核直接判了”reject”,理由是”porn score 92%”。

老张很愤怒:“这是获得过国际摄影奖的作品,你的 AI 说它是色情?”

我手动恢复了老张的图片,但这件事让我意识到:AI 审核不是非黑即白的——它有置信度,有灰色地带。 一个好的审核流程不是”AI 说了算”,而是机器和人的协作。

审核流程架构

我重新设计了审核流程,核心思路是分级审核:

用户上传图片
    │
    ▼
┌─────────────────────┐
│  第一级:AI 快速审核   │  < 300ms
│  (鉴黄 + 二维码)     │
└──────┬──────────────┘
       │
   ┌───┴────┬──────────┐
   │        │          │
   ▼        ▼          ▼
reject    review      pass
   │        │          │
   │        ▼          ▼
   │   ┌──────────┐   异步发起精审
   │   │ 第二级:   │   (OCR + 暴力 + 水印)
   │   │ AI 精审   │
   │   └────┬─────┘
   │        │
   │   ┌────┴─────┐
   │   │          │
   │   ▼          ▼
   │ reject    review
   │   │        │
   │   │        ▼
   │   │   ┌──────────┐
   │   │   │ 第三级:   │
   │   │   │ 人工审核   │  人工审核后台
   │   │   └────┬─────┘
   │   │        │
   │   │   ┌────┴─────┐
   │   │   │          │
   │   │   ▼          ▼
   │   │ reject      pass
   │   │   │          │
   ▼   ▼   ▼          ▼
 拦截通知          用户申诉通道

置信度阈值设计

模型输出的是一个分数,不是一个判决。阈值就是把这个连续分数切成三段动作:

分数区间动作风险
分数大于等于高阈值直接拦截误杀正常内容
分数落在高低阈值之间转人工审核占用人审产能
分数低于低阈值直接放行漏放违规内容

中间那段是人审成本与误杀/漏放风险之间的缓冲带。阈值定得越靠近两端,人审量越少、但误判越极端。

不同风险类型的阈值必须分开,因为它们的代价完全不对称:

风险类型高阈值低阈值设计理由
色情 / 低俗按模型分数设较高值中等高危内容优先保证拦截,误杀可以走申诉
暴恐 / 暴力高中同上,且这类内容几乎没有误杀场景
政治敏感不设直接拦截极低全部转人审,模型判断不能直接决定拦截
二维码 / 广告不设直接拦截中判定为”广告”不等于违规,按业务规则标记或降权
低质内容不设极低只降权不拦截

阈值的来源应该是标注集上的统计,而不是拍脑袋。做法是:用一批人工标注过的图片跑一遍模型,得到每个分数区间的实际误杀率和漏放率,然后按业务约束选点。

约束 1(误杀上限):直接拦截区内,正常图片被判违规的比例 ≤ 可接受上限
约束 2(漏放上限):直接放行区内,违规图片未被拦截的比例 ≤ 可接受上限
约束 3(产能上限):落在人审区间的图片量 ≤ 人审团队能处理的量

满足三条约束的分数点,就是阈值

三条约束同时成立才是合格的阈值。只盯误杀率会漏放,只盯漏放率会误伤,而忽略产能上限则会把人审队列堆到爆。

阈值必须是配置而不是代码。它需要能按场景覆盖(头像更严、营销素材更严、新用户更严),需要能灰度调整,需要在出事当天不发版就能改:

from typing import Optional

# 阈值来自配置表,可以按场景覆盖;调整后灰度生效,不需要发版
DEFAULT_THRESHOLDS = {
    "porn": {"reject": 0.90, "review": 0.55},
    "terror": {"reject": 0.92, "review": 0.60},
    "political": {"reject": None, "review": 0.0},   # 全部转人审
    "qrcode": {"reject": None, "review": 0.80},
}


def decide(
    risk: str,
    score: float,
    thresholds: Optional[dict] = None,
) -> str:
    """把模型分数翻译成动作:reject 拦截 / review 转人审 / pass 放行。

    reject 为 None 表示该风险类型不允许机器直接拦截。
    """
    rule = (thresholds or DEFAULT_THRESHOLDS)[risk]
    reject_at = rule["reject"]
    review_at = rule["review"]

    if reject_at is not None and score >= reject_at:
        return "reject"
    if score >= review_at:
        return "review"
    return "pass"

最后一条纪律:把分数、命中的阈值、模型版本一起写进审核记录。老张那类误判之所以难以复盘,就是因为只留下了”reject”这个结论,没留下”当时分数是多少、用的哪版模型、阈值是多少”。有了这三样,阈值调整才是可回溯、可验证的工程动作,而不是拍脑袋改数字。

人工审核后台

AI 审核不了的图片需要转人工。我设计了一个简单的人工审核后台:

前端审核界面:

用户申诉机制

被拒绝的图片应该有申诉通道:

用户提交申诉
    │
    ▼
POST /api/images/{image_id}/appeals
    │
    ├─ 图片被拒 → 接受申诉,生成申诉单
    ├─ 图片是 pass → 400(没有可申诉的对象)
    └─ 已有进行中的申诉 → 409(避免重复提交刷单)
    │
    ▼
申诉单进入人审队列(优先级高于新上传的图片)
    │
    ▼
人审员看到:原图 + 当时的审核记录(分数、阈值、模型版本)+ 申诉理由
    │
    ├─ 成立 → 恢复图片,通知用户
    └─ 不成立 → 维持拒绝,附上判定依据

申诉接口的设计要点有三个。第一个是申诉必须带上”当时的判定依据”,否则人审员只能凭感觉看图。前面说过要把分数、命中的阈值、模型版本一起写进审核记录——那三样东西在这里派上用场:

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional
from uuid import uuid4


@dataclass
class Appeal:
    id: str
    image_id: str
    user_id: str
    reason: str
    status: str                       # submitted / reviewing / approved / rejected
    submitted_at: datetime
    decided_at: Optional[datetime] = None
    # 申诉时要一并展示的判定依据
    risk_type: Optional[str] = None
    model_score: Optional[float] = None
    threshold_at_decision: Optional[float] = None
    model_version: Optional[str] = None


def submit_appeal(
    image_id: str,
    user_id: str,
    reason: str,
    decision: dict,
    has_open_appeal: bool,
) -> Appeal:
    """创建一条申诉单,并把做出这个决定时的依据一并带过去。"""
    if decision.get("status") != "reject":
        raise ValueError("只有被拒绝的图片才能申诉")
    if has_open_appeal:
        raise ValueError("该图片已有进行中的申诉")

    return Appeal(
        id=f"apl_{uuid4().hex[:12]}",
        image_id=image_id,
        user_id=user_id,
        reason=reason,
        status="submitted",
        submitted_at=datetime.now(timezone.utc),
        risk_type=decision.get("risk_type"),
        model_score=decision.get("score"),
        threshold_at_decision=decision.get("threshold"),
        model_version=decision.get("model_version"),
    )

第二个是申诉状态机。状态不能随便跳,每一步都要留下时间和操作人:

状态含义进入条件谁能操作
submitted已提交,等待分配用户提交申诉用户
reviewing人审员正在看从队列领单人审员
approved申诉成立,图片恢复人审判定误杀人审员
rejected申诉不成立,维持拒绝人审判定判定正确人审员

第三个是申诉结论要能被发现。用户不会盯着后台等结果,所以申诉结论要通过站内信或者邮件推送给用户;图片被恢复的,还要同步刷新 CDN 缓存——否则边缘还在返回被拒时的占位图或者旧缓存,用户刷新页面还是看不到。

申诉成立
   │
   ├─→ 数据库:图片状态改回 visible
   ├─→ 审核记录:标记为"误杀",进入阈值调优的样本池
   ├─→ CDN:刷新该图片的所有派生图缓存
   └─→ 通知:站内信 + 邮件,告诉用户申诉结果和原因

申诉不只是”给用户一个交代”,它同时是阈值调优的数据来源。申诉成立率是审核看板上的关键指标(见”审核数据看板”):如果某个风险类型的申诉成立率突然升高,说明阈值定得太激了;如果接近 0,人审可能只是在批量点”通过”。

有两个坑要提前堵上:

  1. 申诉入口要限流。否则会出现”被拒就申诉”的自动化刷单,申诉队列被灌满,真正需要人审的用户反而排不上。同一张图只允许一条进行中的申诉,用户维度的申诉频率也要设上限。
  2. 申诉结论要写理由。“审核不通过”对用户没有任何信息量。“经复核,该图片属于艺术人体摄影,符合社区规范”才有意义。用户理解了规则,才会减少重复申诉。

灰度发布审核规则

审核规则和阈值的调整需要灰度发布——先在小比例图片上验证效果,再全量上线:

新规则上线(调整阈值 / 换模型版本 / 加新审核维度)

    阶段 1:影子模式
    新规则和旧规则同时跑,只记录新规则的判定结果,不影响用户
    观察:新旧判定差异率、差异样本的质量
    ─────────────────────────────────────────────
    阶段 2:小流量灰度(1%)
    按图片 ID 哈希分桶,命中灰度桶的走新规则,其余走旧规则
    观察:误杀率、申诉量、人审量
    ─────────────────────────────────────────────
    阶段 3:扩大灰度(10% → 50% → 100%)
    每一步都设一个"停留期",指标稳定才继续
    ─────────────────────────────────────────────
    任一阶段指标恶化 → 回滚(改配置,不发版)

影子模式是最重要的一步。审核规则的改动最容易出事故的地方,是它直接影响用户能不能看到自己的作品。影子模式让新规则”只说不做”:判定结果照常记录,但真正生效的仍然是旧规则。这样可以在零风险的前提下,拿到新旧规则的差异率,也能让人提前看一遍差异样本长什么样。

分桶的实现要稳定。同一个图片 ID,在任何时候、任何机器上,都必须落到同一个桶,否则同一张图会一会儿被拦一会儿不被拦:

import hashlib


def bucket_of(image_id: str, buckets: int = 100) -> int:
    """按图片 ID 哈希分桶,返回 0 到 buckets-1 的桶号。

    用图片 ID 而不是用户 ID 分桶:同一张图对所有人应该是同一个判定结果,
    否则会出现"我看不到你却能看见"的诡异现象。
    """
    digest = hashlib.sha256(image_id.encode("utf-8")).hexdigest()
    return int(digest[:8], 16) % buckets


def should_use_new_rule(image_id: str, rollout_percent: int) -> bool:
    """灰度开关。比例为 0 时全走旧规则,100 时全走新规则。"""
    if rollout_percent >= 100:
        return True
    if rollout_percent <= 0:
        return False
    return bucket_of(image_id) < rollout_percent

每个阶段要盯的指标和回滚条件:

阶段流量比例核心观察指标回滚条件
影子100%(不影响判定)新旧判定差异率、差异样本分布差异率超出预期,先停下来分析
灰度一1%申诉量、人审队列长度申诉量或人审量突增
灰度二10%误杀率、改判率改判率明显高于历史基线
灰度三50%同上,加 P99 审核延迟延迟上升影响用户可见速度
全量100%全部看板指标保留一键回滚开关至少一周

两条纪律:

  1. 回滚必须只需要改配置。前面说过”阈值必须是配置而不是代码”——灰度发布能做到一键回滚,前提是规则和阈值本来就在配置里。如果规则写在代码里,回滚就意味着发版,而半夜发版是不可靠的。
  2. 每次灰度都要留下”改了什么”的记录:规则版本号、阈值、模型版本、生效时间、灰度比例。有了这个记录,多个灰度并行时才能回答”这张图当时是按哪版规则判的”。

灰度期间还有一个容易被忽略的动作:老张那类误判样本要单独看。人审改判的样本是最好的负样本,把它们单独拉出来看,比看总体指标更容易发现新规则的问题在哪。

审核数据看板

审核系统是一个”花钱买安全”的系统:机审调用要钱,人审要工时,误杀要赔用户信任。所以它的看板不能只看”处理了多少”,必须能看出质量、产能和时效三件事。

模块指标口径关注什么
总量上传量、审核完成量、队列积压量事件计数审核速度跟不跟得上上传速度
时效从上传到可见的 P50 / P99、平均处理时长事件时间戳差值用户等得起等不起
判定分布pass / review / reject 占比、分类命中分布审核记录拦截率是否异常波动
人审产能队列长度、平均等待时长、人均处理量人审队列记录人审是不是堵住了
人审改判率人审结论与机审结论不一致的比例两条记录比对机审阈值是否还合适
申诉申诉量、申诉成立率、平均处理时长申诉记录误报的真实代价有多大
模型表现分模型版本的拦截率、分数分布审核记录新模型上线后是否退化
成本机审调用量、人审工时账单 + 记录单张审核成本是否失控

其中人审改判率是整套看板里最有价值的一个指标:

  • 改判率高,说明机审阈值偏激或者模型在这批内容上不准,该调阈值或补样本;
  • 改判率接近 0,要警惕两种可能——要么机审确实很准,要么人审在批量点”通过”没有认真看。区分办法是抽样复核人审自己的准确率。

看板最终要能回答一个具体问题:“这张图为什么被拒?” 对应的查询链路是:图片 ID → 审核记录 → 命中的风险类型 → 模型分数 → 当时生效的阈值 → 模型版本 → 命中了哪条规则。老张那类误判之所以难复盘,就是因为这条链路上缺了”分数”和”阈值”两环;补齐之后,阈值调整才从”拍脑袋”变成”有据可依”。

配套的告警:

告警触发条件为什么要紧
队列积压审核队列长度持续超过阈值直接影响用户看到图片的时长
拦截率突变日拦截率相对基线大幅波动可能是模型故障,也可能是攻击性批量上传
申诉激增申诉量突增大概率是阈值或模型出了问题
机审失败调用失败率上升失败被当成放行,就是最危险的漏放

最后强调一遍看板上数据的意义:审核看板的第一读者是当天值班的人。它的设计目标不是”展示全面”,而是”出问题时,三分钟内能定位到是哪一环、该做什么”。

发现内容有问题?

告诉我们哪里需要改进。