审核流程设计

一次误判引发的”事故”

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

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

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

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

审核流程架构

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

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

置信度阈值设计

from dataclasses import dataclass
from typing import Optional

@dataclass
class AuditThreshold:
    """审核阈值配置"""
    # 鉴黄阈值
    porn_reject: float = 90.0     # > 90% 直接拒绝
    porn_review: float = 60.0     # 60~90% 人工审核
    
    # 暴力阈值
    violence_reject: float = 85.0
    violence_review: float = 50.0
    
    # 性感内容
    sexy_reject: float = 95.0     # 性感内容不轻易拒绝（泳装等）
    sexy_review: float = 70.0
    
    # 文字风险
    text_reject: str = 'high'     # high/medium/low
    text_review: str = 'medium'
    
    # 是否需要人工确认 AI 的 reject 决定
    human_confirm_reject: bool = False  # 摄影社区建议开启


class AuditDecisionEngine:
    """审核决策引擎"""
    
    def __init__(self, thresholds: AuditThreshold = None):
        self.thresholds = thresholds or AuditThreshold()
    
    def decide(self, audit_results: dict) -> dict:
        """综合所有审核结果，做出最终决策"""
        decisions = []
        
        for dimension, result in audit_results.items():
            decision = self._decide_dimension(dimension, result)
            decisions.append(decision)
        
        # 取最严格的决定
        action_priority = {'reject': 3, 'review': 2, 'pass': 1}
        final = max(decisions, key=lambda d: action_priority.get(d['action'], 0))
        
        return {
            'action': final['action'],
            'dimensions': decisions,
            'primary_reason': final.get('reason'),
            'confidence': final.get('confidence', 0),
        }
    
    def _decide_dimension(self, dimension: str, result: dict) -> dict:
        """单个维度的决策"""
        confidence = result.get('confidence', 0) or result.get('rate', 0)
        label = result.get('label', 'normal')
        
        if dimension == 'nsfw':
            if label == 'porn':
                if confidence >= self.thresholds.porn_reject:
                    return {'action': 'reject', 'confidence': confidence,
                            'reason': f'色情内容 (置信度 {confidence:.0f}%)'}
                elif confidence >= self.thresholds.porn_review:
                    return {'action': 'review', 'confidence': confidence,
                            'reason': f'疑似色情 (置信度 {confidence:.0f}%)'}
            elif label == 'sexy':
                if confidence >= self.thresholds.sexy_reject:
                    return {'action': 'reject', 'confidence': confidence,
                            'reason': f'性感内容 (置信度 {confidence:.0f}%)'}
                elif confidence >= self.thresholds.sexy_review:
                    return {'action': 'review', 'confidence': confidence,
                            'reason': f'疑似性感 (置信度 {confidence:.0f}%)'}
        
        elif dimension == 'violence':
            if confidence >= self.thresholds.violence_reject:
                return {'action': 'reject', 'confidence': confidence,
                        'reason': f'暴力内容 (置信度 {confidence:.0f}%)'}
            elif confidence >= self.thresholds.violence_review:
                return {'action': 'review', 'confidence': confidence,
                        'reason': f'疑似暴力 (置信度 {confidence:.0f}%)'}
        
        elif dimension == 'text':
            severity = result.get('severity', 'none')
            if severity == 'high':
                return {'action': 'reject', 'confidence': 90,
                        'reason': f'敏感文字: {result.get("word", "")}'}
            elif severity == 'medium':
                return {'action': 'review', 'confidence': 70,
                        'reason': f'可疑文字: {result.get("word", "")}'}
        
        return {'action': 'pass', 'confidence': confidence, 'reason': None}

人工审核后台

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

# 人工审核后台 API
from flask import Blueprint, request, jsonify

audit_bp = Blueprint('audit_admin', __name__)

@audit_bp.route('/api/admin/audit/pending', methods=['GET'])
def get_pending_reviews():
    """获取待人工审核的图片列表"""
    page = request.args.get('page', 1, type=int)
    per_page = 20
    
    photos = db.query_paginated('photos', {
        'status': 'pending_review',
    }, page=page, per_page=per_page, order_by='created_at ASC')
    
    results = []
    for photo in photos:
        audit = db.query('photo_audits', {'photo_id': photo['id']})
        results.append({
            'photo_id': photo['id'],
            'image_url': f"https://cdn.guangying.com/{photo['object_key']}",
            'ai_decision': audit['action'],
            'ai_reason': audit.get('review_reason', ''),
            'ai_scores': audit.get('scores', {}),
            'uploader': {
                'id': photo['user_id'],
                'name': photo.get('username', ''),
                'is_verified': photo.get('is_verified', False),
            },
            'uploaded_at': photo['created_at'].isoformat(),
        })
    
    return jsonify({
        'items': results,
        'total': db.count('photos', {'status': 'pending_review'}),
        'page': page,
    })


@audit_bp.route('/api/admin/audit/<photo_id>/decision', methods=['POST'])
def human_decision(photo_id):
    """人工审核决定"""
    data = request.json
    decision = data['decision']  # 'approve' or 'reject'
    reason = data.get('reason', '')
    reviewer_id = get_current_admin_id()
    
    photo = db.query('photos', {'id': photo_id})
    if not photo:
        return jsonify({'error': 'not found'}), 404
    
    if decision == 'approve':
        # 放行
        db.update('photos', {'status': 'approved'}, {'id': photo_id})
        
        # 触发后续处理（压缩、CDN 等）
        queue.publish('image:process', {
            'photo_id': photo_id,
            'object_key': photo['object_key'],
            'bypass_audit': True,
        })
        
        action = 'human_approve'
    
    elif decision == 'reject':
        # 拒绝
        db.update('photos', {'status': 'rejected'}, {'id': photo_id})
        
        # 从 OSS 和 CDN 移除
        oss_client.delete_object(photo['object_key'])
        cdn_client.refresh_object_caches(
            f"https://cdn.guangying.com/{photo['object_key']}"
        )
        
        action = 'human_reject'
    
    # 记录审核日志
    db.insert('audit_logs', {
        'photo_id': photo_id,
        'action': action,
        'reason': reason,
        'reviewer_id': reviewer_id,
        'reviewed_at': datetime.now(),
    })
    
    # 通知用户
    notify_user(photo['user_id'], photo_id, action, reason)
    
    return jsonify({'status': 'ok'})

前端审核界面：

// 人工审核后台界面
interface AuditItem {
  photo_id: string;
  image_url: string;
  ai_decision: string;
  ai_reason: string;
  uploader: { id: number; name: string; is_verified: boolean };
  uploaded_at: string;
}

function AuditDashboard() {
  const [items, setItems] = useState<AuditItem[]>([]);
  const [currentIndex, setCurrentIndex] = useState(0);

  // 加载待审核列表
  useEffect(() => {
    fetch('/api/admin/audit/pending')
      .then(r => r.json())
      .then(data => setItems(data.items));
  }, []);

  const handleDecision = async (decision: 'approve' | 'reject') => {
    const item = items[currentIndex];
    
    await fetch(`/api/admin/audit/${item.photo_id}/decision`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        decision,
        reason: decision === 'reject' ? prompt('请输入拒绝原因') : '',
      }),
    });
    
    // 下一个
    setCurrentIndex(prev => prev + 1);
  };

  const item = items[currentIndex];
  if (!item) return <div>暂无待审核图片</div>;

  return (
    <div className="audit-dashboard">
      <div className="image-preview">
        <img src={item.image_url} alt="审核图片" />
      </div>
      
      <div className="audit-info">
        <p>AI 判定: {item.ai_decision} - {item.ai_reason}</p>
        <p>上传者: {item.uploader.name} 
          {item.uploader.is_verified && ' ✓认证'}
        </p>
      </div>
      
      <div className="audit-actions">
        <button onClick={() => handleDecision('approve')}>
          ✅ 通过
        </button>
        <button onClick={() => handleDecision('reject')}>
          ❌ 拒绝
        </button>
      </div>
      
      {/* 键盘快捷键 */}
      <KeyboardHandler
        onKeyA={() => handleDecision('approve')}
        onKeyR={() => handleDecision('reject')}
      />
    </div>
  );
}

用户申诉机制

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

@audit_bp.route('/api/appeal', methods=['POST'])
def submit_appeal():
    """用户提交申诉"""
    data = request.json
    photo_id = data['photo_id']
    reason = data['reason']
    user_id = get_current_user_id()
    
    photo = db.query('photos', {'id': photo_id, 'user_id': user_id})
    if not photo:
        return jsonify({'error': 'not found'}), 404
    
    if photo['status'] != 'rejected':
        return jsonify({'error': '只能申诉被拒绝的图片'}), 400
    
    # 检查申诉频率（每用户每天最多 3 次）
    today = datetime.now().date()
    appeal_count = db.count('appeals', {
        'user_id': user_id,
        'created_at__gte': today,
    })
    if appeal_count >= 3:
        return jsonify({'error': '今日申诉次数已达上限'}), 429
    
    # 创建申诉
    appeal_id = db.insert('appeals', {
        'photo_id': photo_id,
        'user_id': user_id,
        'reason': reason,
        'status': 'pending',
        'created_at': datetime.now(),
    })
    
    # 通知高级审核员（申诉由更高级别的审核员处理）
    queue.publish('appeal:review', {
        'appeal_id': appeal_id,
        'photo_id': photo_id,
    })
    
    return jsonify({'appeal_id': appeal_id, 'status': 'pending'})


@audit_bp.route('/api/appeal/<appeal_id>/resolve', methods=['POST'])
def resolve_appeal(appeal_id):
    """高级审核员处理申诉"""
    data = request.json
    decision = data['decision']  # 'uphold' or 'overturn'
    reason = data.get('reason', '')
    
    appeal = db.query('appeals', {'id': appeal_id})
    photo_id = appeal['photo_id']
    
    if decision == 'overturn':
        # 翻案：恢复图片
        db.update('photos', {'status': 'approved'}, {'id': photo_id})
        db.update('appeals', {'status': 'overturn', 'resolved_at': datetime.now()}, {'id': appeal_id})
        
        # 触发图片处理
        photo = db.query('photos', {'id': photo_id})
        queue.publish('image:process', {
            'photo_id': photo_id,
            'object_key': photo['object_key'],
            'bypass_audit': True,
        })
        
        # 用这次申诉的样本反馈给 AI 模型（提升准确率）
        feedback_to_model(photo_id, 'should_pass')
    
    else:
        # 维持原判
        db.update('appeals', {'status': 'uphold', 'resolved_at': datetime.now()}, {'id': appeal_id})
        
        feedback_to_model(photo_id, 'should_reject')
    
    return jsonify({'status': 'ok'})

灰度发布审核规则

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

class AuditABTest:
    """审核规则 A/B 测试"""
    
    def __init__(self):
        self.experiments = {
            'new_porn_threshold': {
                'control': {'porn_reject': 90.0, 'porn_review': 60.0},
                'treatment': {'porn_reject': 85.0, 'porn_review': 55.0},
                'traffic_pct': 10,  # 10% 流量走新规则
                'enabled': True,
            },
        }
    
    def get_thresholds(self, photo_id: str) -> AuditThreshold:
        """获取当前图片应该使用的阈值"""
        for exp_name, config in self.experiments.items():
            if not config['enabled']:
                continue
            
            # 基于 photo_id 做流量分配
            bucket = int(hashlib.md5(photo_id.encode()).hexdigest()[:4], 16) % 100
            
            if bucket < config['traffic_pct']:
                # 走实验组
                return AuditThreshold(**config['treatment'])
            else:
                # 走对照组
                return AuditThreshold(**config['control'])
        
        return AuditThreshold()  # 默认阈值
    
    def record_result(self, photo_id: str, ai_action: str, human_action: str):
        """记录 AI 和人工判定结果，用于评估实验效果"""
        for exp_name, config in self.experiments.items():
            if not config['enabled']:
                continue
            
            bucket = int(hashlib.md5(photo_id.encode()).hexdigest()[:4], 16) % 100
            group = 'treatment' if bucket < config['traffic_pct'] else 'control'
            
            db.insert('audit_experiments', {
                'experiment': exp_name,
                'photo_id': photo_id,
                'group': group,
                'ai_action': ai_action,
                'human_action': human_action,
                'agree': ai_action == human_action,
                'created_at': datetime.now(),
            })
    
    def evaluate(self, experiment_name: str):
        """评估实验效果"""
        results = db.query_all('audit_experiments', {'experiment': experiment_name})
        
        groups = {'control': [], 'treatment': []}
        for r in results:
            groups[r['group']].append(r)
        
        for group_name, data in groups.items():
            total = len(data)
            agreed = sum(1 for d in data if d['agree'])
            false_positives = sum(1 for d in data 
                                 if d['ai_action'] == 'reject' and d['human_action'] != 'reject')
            false_negatives = sum(1 for d in data 
                                 if d['ai_action'] != 'reject' and d['human_action'] == 'reject')
            
            print(f"\n{group_name}:")
            print(f"  AI与人工一致率: {agreed/total*100:.1f}%")
            print(f"  误报率: {false_positives/total*100:.1f}%")
            print(f"  漏报率: {false_negatives/total*100:.1f}%")

审核数据看板

class AuditDashboard:
    """审核数据统计"""
    
    def get_stats(self, date_range: tuple) -> dict:
        """获取审核统计数据"""
        start, end = date_range
        
        total = db.count('photos', {'created_at__between': (start, end)})
        
        by_status = {}
        for status in ['approved', 'rejected', 'pending_review', 'processing']:
            by_status[status] = db.count('photos', {
                'status': status,
                'created_at__between': (start, end),
            })
        
        by_dimension = {}
        for dim in ['nsfw', 'violence', 'text', 'qrcode', 'watermark']:
            by_dimension[dim] = db.count('photo_audits', {
                'primary_dimension': dim,
                'action': 'reject',
                'created_at__between': (start, end),
            })
        
        # 人工审核统计
        human_stats = {
            'total_reviewed': db.count('audit_logs', {'created_at__between': (start, end)}),
            'overturn_rate': self._overturn_rate(start, end),
            'avg_review_time': self._avg_review_time(start, end),
        }
        
        return {
            'period': {'start': start.isoformat(), 'end': end.isoformat()},
            'total_uploads': total,
            'status_breakdown': by_status,
            'rejection_by_dimension': by_dimension,
            'human_review': human_stats,
            'auto_reject_rate': by_status.get('rejected', 0) / max(total, 1) * 100,
        }

本节小结

✅ 我学到了什么：

• 审核流程不是”AI 一刀切”，而是机审→人审的分级体系
• 置信度阈值是平衡误报和漏报的关键——摄影社区应该偏保守
• 灰度发布审核规则，避免一次调参影响全量用户
• 用户申诉机制是纠错的最后保障，申诉数据可以反哺 AI 模型

⚠️ 踩过的坑：

• AI 审核对人体艺术的误判率很高——这类内容必须走人工审核
• 审核阈值不能太严格，15% 的误报率会让摄影师愤怒地离开平台
• 人工审核的效率瓶颈——高峰期可能积压大量待审核图片

🎯 下一步：审核流程完善了，图片处理也搞定了。但用户访问图片还是从我的一台服务器上下载。是时候上 CDN 了。

我的思考

思考 1

如果人工审核员也有主观偏差（比如对同一张图，A 审核员通过了，B 审核员拒绝了），怎么保证审核标准的一致性？

建立详细的审核指南文档：
- 什么是色情？什么是艺术？
- 具体案例 + 判定结果
- 边界案例的处理原则

定期用标注数据集测试审核员：
- 准备 100 张标准答案已知的图片
- 审核员判定结果与标准答案对比
- 一致率低于 90% 的需要培训

# 高风险内容由两个审核员独立审核
def dual_review(photo_id):
    # 两个审核员独立给出判定
    review_1 = get_review_from_reviewer(photo_id, reviewer_pool[0])
    review_2 = get_review_from_reviewer(photo_id, reviewer_pool[1])
    
    if review_1.decision == review_2.decision:
        # 一致，采用该判定
        return review_1.decision
    else:
        # 不一致，升级到高级审核员仲裁
        return escalate_to_senior(photo_id, review_1, review_2)

每周一次审核校准会议：
- 回顾本周的争议案例
- 讨论审核标准的边界
- 更新审核指南
- 统一团队认知

# 检测审核员的偏差
def detect_reviewer_bias(reviewer_id, period='30d'):
    decisions = get_reviewer_decisions(reviewer_id, period)
    
    # 对比该审核员与团队平均的偏差
    team_avg_reject_rate = get_team_avg('reject_rate')
    my_reject_rate = decisions['reject'] / decisions['total']
    
    if abs(my_reject_rate - team_avg_reject_rate) > 0.15:
        # 偏差超过 15%，标记为异常
        alert(f"审核员 {reviewer_id} 的拒绝率偏差过大")

思考 2

对于”先审后发”（审核通过才展示）和”先发后审”（先展示，发现问题再删除），你会怎么选择？各有什么利弊？

先审后发（审核通过才展示）：
  优点：
    - 平台零违规风险
    - 用户体验一致（看到的都是合规内容）
  
  缺点：
    - 上传到展示有延迟（等待审核，10秒~数分钟）
    - 审核高峰期可能积压
    - 摄影师即时分享的体验被破坏

先发后审（先展示后审核）：
  优点：
    - 即时展示，用户体验好
    - 审核压力小（可以慢慢审）
  
  缺点：
    - 违规内容可能短暂可见
    - 法律风险
    - 用户可能看到不当内容

def choose_audit_strategy(user):
    """根据用户风险等级选择审核策略"""
    
    risk = calculate_user_risk(user)
    
    if risk == 'high':
        # 新用户、曾被拒绝过的用户 → 先审后发
        return 'pre_audit'
    
    elif risk == 'medium':
        # 普通用户 → 先发后审，但降低展示范围
        return 'post_audit_limited'
    
    else:
        # 认证摄影师、长期无违规的用户 → 先发后审
        return 'post_audit'


def calculate_user_risk(user):
    days_since_register = (datetime.now() - user.created_at).days
    total_uploads = user.total_uploads
    rejected_count = user.rejected_count
    reject_rate = rejected_count / max(total_uploads, 1)
    
    if days_since_register < 7:
        return 'high'
    if reject_rate > 0.1:
        return 'high'
    if reject_rate > 0.02:
        return 'medium'
    if user.is_verified and reject_rate == 0:
        return 'low'
    return 'medium'