当前位置：首页 > news >正文

内容审核流水线：文本、图片和视频的异步审核架构设计

news 2026/7/22 2:03:52

内容审核流水线：文本、图片和视频的异步审核架构设计

一、多模态内容爆发下的审核架构困局

UGC平台日均内容发布量突破百万级别后，同步审核模式的瓶颈会直接暴露在生产监控面板上。一次同步审核请求串行调用文本敏感词匹配、图片鉴黄模型和视频关键帧抽取，端到端耗时落在 800ms 到 2s 之间。当上游业务方的 HTTP 超时设为 1s 时，超过一半的审核请求直接超时丢弃——这意味着该被拦截的内容漏过去了。

问题不在于单一模型太慢，而在于耦合。文本、图片、视频三种模态的审核逻辑被塞进同一个请求-响应周期里，慢的环节拖死快的环节。更致命的是，视频审核需要先做关键帧抽取再做逐帧推理，天然无法在 1s 内完成。

基础设施不需要漂亮话，需要的是把三种模态的审核拆成独立流水线、各自异步执行的架构。

二、流水线解耦：消息队列驱动的多模态异步架构

核心思路是把"审核"这件事从同步接口调用变成任务提交-结果回调的异步模型。业务方发布内容后，审核服务只做两件事：接收内容元数据、将审核任务投递到对应的消息队列，然后立即返回任务 ID。

每种模态对应一条独立的消费流水线。

文本审核流水线按优先级分为两级：第一级是 AC 自动机做敏感词匹配，耗时通常 5ms 以内；第二级是 NLP 模型做语义级别的违规识别，耗时 50ms-200ms。两级之间通过同一个队列的不同 Consumer Group 串联——第一级判定违规的直接终止流水线，只有疑似或通过的内容才进入第二级。

图片审核的挑战在于图片本身不跟请求体一起提交。业务方传的是 CDN URL，Worker 需要先下载图片再做推理。这里不能阻塞在下载上——单 Worker 用同步 HTTP GET 下载图片的 P99 延迟可能到 2s。做法是每个 Worker 内部用 goroutine 池并发下载多张图片，下载超时设为 500ms，超时直接标记为"审核失败-下载超时"。

视频审核是三层漏斗：关键帧抽取 → 图片审核（复用图片流水线）→ 音频转文字审核。关键帧抽取用 FFmpeg 按 I 帧间隔抓帧，10 分钟视频约产生 600 帧，抽帧后逐帧投递到图片审核队列，最后汇总所有帧的审核结果取最严判定。

三、Go 生产级任务路由与 Worker Pool 实现

任务路由的核心是把请求中的内容类型映射到对应的消息队列 Topic 上。下面是一个简化但可投产的实现思路。

type ModerationTask struct { TaskID string `json:"task_id"` BizID string `json:"biz_id"` MediaType string `json:"media_type"` // text/image/video Content string `json:"content"` // text or URL Callback string `json:"callback"` CreatedAt time.Time `json:"created_at"` } // RouteTask 根据媒体类型将任务路由到对应队列 func (r *TaskRouter) RouteTask(ctx context.Context, task ModerationTask) error { var topic string switch task.MediaType { case "text": topic = "moderation.text" case "image": topic = "moderation.image" case "video": topic = "moderation.video" default: return fmt.Errorf("unsupported media type: %s", task.MediaType) } payload, err := json.Marshal(task) if err != nil { return fmt.Errorf("marshal task: %w", err) } // RabbitMQ 投递，消息持久化 + 手动确认 return r.ch.PublishWithContext(ctx, "moderation.exchange", // exchange topic, // routing key true, // mandatory false, // immediate amqp.Publishing{ DeliveryMode: amqp.Persistent, ContentType: "application/json", Body: payload, MessageId: task.TaskID, }, ) }

Worker Pool 侧的关键设计是并发度与 QPS 的匹配。假设单 Worker 处理一条文本审核耗时 10ms，期望吞吐 1000 QPS，理论上 10 个 goroutine 就够了。但实际中网络抖动、GC 停顿会让 P99 膨胀到 30ms，所以需要按 P99 延迟来算并发数：concurrency = target_qps * p99_latency，即1000 * 0.03 = 30。预留 1.5 倍余量，实际开 45 个 goroutine。

type WorkerPool struct { tasks <-chan ModerationTask handler TaskHandler workers int } func (wp *WorkerPool) Start(ctx context.Context) { for i := 0; i < wp.workers; i++ { go func(workerID int) { for { select { case <-ctx.Done(): return case task, ok := <-wp.tasks: if !ok { return } // 每条消息独立超时控制，不因单条慢消息阻塞整个 Worker taskCtx, cancel := context.WithTimeout(ctx, 30*time.Second) if err := wp.handler.Handle(taskCtx, task); err != nil { log.Printf("worker %d handle task %s: %v", workerID, task.TaskID, err) } cancel() } } }(i) } }