内容审核 Agent：多模态内容安全检测系统的 RAG 增强方案

内容审核 Agent：多模态内容安全检测系统的 RAG 增强方案

一、深度引言与场景痛点

大家好，我是赵咕咕。

去年我们接手了一个内容审核的项目，需求听起来很直接：用大模型替代人工审核员，对 UGC 内容做安全合规检测。文本、图片、视频片段，三模态混合输入，输出一个 pass/block/review 的决策。

一开始团队信心满满——GPT-4V 都出来了，多模态审核还不是调个 API 的事？结果第一版上线三天就出了事故：一张包含"裸体雕像艺术照"被标记为违规，一幅"包含血液的医学解剖图"被判为色情。而真正需要拦截的——用谐音字和换行符绕过敏感词检测的违规文本——却被干干净净地放行了。

这说明一个问题：纯 LLM 审核不具备领域知识。它不知道你的平台对"泳装"是容忍还是不通过，也不知道哪些行业术语（比如医疗解剖）需要白名单放行。而这些问题，恰好是 RAG 的强项——它能为 LLM 注入你业务场景下的审核标准、违规案例库和白名单规则。

这篇文章，我把多模态审核系统从"调 API"到"RAG 增强"的完整工程方案整理出来。

二、底层机制与原理深度剖析

2.1 为什么纯 LLM 审核不够？

纯 LLM 做内容审核有几个致命的短板：

• 上下文缺失：LLM 不知道你的平台规范。同样是"枪"，游戏论坛和军事论坛的审核标准完全不同。
• 幻觉风险：LLM 可能"脑补"违规内容。一张正常的医学图片，它可能因为"看到红色液体"而判断为暴力。
• 对抗样本脆弱：用户用谐音、拆字、Unicode 混淆绕过检测，LLM 可能被欺骗。

而 RAG 能解决这些问题：把审核规范、历史判例、违规黑名单、白名单规则做向量化索引，审核时检索最相关的判例作为参考上下文。

2.2 多模态审核流水线架构

这个流水线的关键设计点：

1. 多模态 pipeline 解耦：文本、图片、视频分别走各自的 embedding 管道，最后在向量检索层统一。这样新增模态不需要改核心审核逻辑。
2. 规则库 + 向量检索双路：向量检索负责"找相似案例"，规则引擎负责"硬匹配"。比如手机号正则检测不需要过 LLM，直接用规则拦截。
3. 反馈闭环：人工复审的结果会写回向量库，形成审核案例的持续积累。

2.3 RAG 在审核中的角色

RAG 在这里做两件事：

正向增强：检索历史判例。当 LLM 审核一条"包含手术视频片段"的内容时，检索到之前人工确认过的"医学手术内容-通过"的案例，LLM 就会降低误判概率。

负向增强：检索违规案例库。当 LLM 看到一条文本包含"代办信用卡"（用分隔符绕过检测），检索到历史上类似案例被标记为"违规-金融诈骗"，LLM 就能识别出这是对抗样本。

核心原理就是把人工审核员的经验和平台规范变成了可检索的向量知识。

三、生产级代码实现

下面给出一个支持文本+图片双模态的生产级审核 Agent 实现：

import asyncio import logging import base64 from dataclasses import dataclass from enum import Enum from typing import Any import numpy as np from openai import AsyncOpenAI from pydantic import BaseModel, Field logger = logging.getLogger(__name__) class AuditDecision(str, Enum): PASS = "pass" BLOCK = "block" REVIEW = "review" @dataclass class AuditCase: """历史审核案例。""" case_id: str content_text: str content_type: str # text / image / video decision: AuditDecision reason: str embedding: np.ndarray | None = None class ContentInput(BaseModel): text: str = "" image_base64: str | None = None content_type: str = "text" class AuditResult(BaseModel): decision: AuditDecision confidence: float = Field(ge=0.0, le=1.0) reason: str references: list[str] = Field(default_factory=list) class MultimodalAuditAgent: """多模态内容审核 Agent（RAG 增强版）。""" def __init__( self, llm_client: AsyncOpenAI, vector_store: Any, # FAISS / Milvus 客户端 embedding_model: str = "bge-large-zh-v1.5", top_k: int = 5, similarity_threshold: float = 0.75, ): self._llm = llm_client self._vector_store = vector_store self._embedding_model = embedding_model self._top_k = top_k self._threshold = similarity_threshold # 硬规则：不走 LLM，直接拦截 self._blocklist_patterns: list[tuple[str, str]] = [ ("phone", r"1[3-9]\d{9}"), ("idcard", r"[1-9]\d{5}(19|20)\d{10}[\dXx]"), ] async def audit( self, content: ContentInput, timeout: float = 30.0 ) -> AuditResult: """对内容做审核，返回决策结果。""" try: return await asyncio.wait_for( self._audit_impl(content), timeout=timeout ) except asyncio.TimeoutError: logger.error("审核超时，降级为人工复审") return AuditResult( decision=AuditDecision.REVIEW, confidence=0.0, reason="审核超时，转人工处理", ) async def _audit_impl(self, content: ContentInput) -> AuditResult: # ── 第零步：硬规则快速拦截 ── hard_rule_hit = self._check_hard_rules(content.text) if hard_rule_hit: return AuditResult( decision=AuditDecision.BLOCK, confidence=1.0, reason=hard_rule_hit, ) # ── 第一步：构建内容 embedding ── content_emb = await self._embed_content(content) # ── 第二步：检索相似历史案例 ── similar_cases = await self._retrieve_cases(content_emb) # ── 第三步：白名单检查 ── if self._check_whitelist(content.text, similar_cases): return AuditResult( decision=AuditDecision.PASS, confidence=0.95, reason="命中白名单规则", ) # ── 第四步：LLM 综合判断 ── result = await self._llm_judge(content, similar_cases) return result def _check_hard_rules(self, text: str) -> str | None: """硬规则检测：手机号、身份证等直接拦截。""" import re for label, pattern in self._blocklist_patterns: if re.search(pattern, text): return f"检测到敏感信息: {label}，直接拦截" return None def _check_whitelist( self, text: str, cases: list[AuditCase] ) -> bool: """白名单检查：历史案例中有高相似度的 PASS 案例则放行。""" for case in cases: if ( case.decision == AuditDecision.PASS and self._text_overlap(text, case.content_text) > 0.8 ): return True return False @staticmethod def _text_overlap(a: str, b: str) -> float: """简单文本重叠度计算（生产环境建议用 jaccard）。""" set_a = set(a) set_b = set(b) if not set_a or not set_b: return 0.0 return len(set_a & set_b) / min(len(set_a), len(set_b)) async def _embed_content( self, content: ContentInput ) -> np.ndarray: """构建内容 embedding。文本走文本模型，图片走 CLIP。""" if content.image_base64 and content.content_type == "image": # 图片走 CLIP 多模态 embedding try: response = await self._llm.embeddings.create( model="clip-vit-large-patch14", input=content.text or "image content", ) except Exception as e: logger.warning("图片 embedding 失败: %s，降级为文本", e) response = await self._llm.embeddings.create( model=self._embedding_model, input=content.text or "unknown image", ) else: response = await self._llm.embeddings.create( model=self._embedding_model, input=content.text, ) emb = np.array(response.data[0].embedding, dtype=np.float32) return emb async def _retrieve_cases( self, query_emb: np.ndarray ) -> list[AuditCase]: """向量检索相似历史审核案例。""" try: results = self._vector_store.search( query_emb.tolist(), top_k=self._top_k ) cases = [] for r in results: if r.get("score", 0) >= self._threshold: cases.append(AuditCase( case_id=r["case_id"], content_text=r.get("text", ""), content_type=r.get("type", "text"), decision=AuditDecision(r.get("decision", "review")), reason=r.get("reason", ""), )) return cases except Exception as e: logger.error("向量检索失败: %s，降级为零案例审核", e) return [] async def _llm_judge( self, content: ContentInput, cases: list[AuditCase] ) -> AuditResult: """LLM 结合检索案例做最终判断。""" # 构建参考案例文本 case_texts = [] for i, c in enumerate(cases, 1): case_texts.append( f"案例{i}: 内容={c.content_text[:100]}, " f"判定={c.decision.value}, 理由={c.reason}" ) references = "\n".join(case_texts) if case_texts else "无相关历史案例" prompt = f"""你是内容审核专家。请审核以下内容并给出判断。 审核规范： - 禁止色情、暴力、违法、诈骗内容 - 医疗、教育、艺术类内容正常放行 - 不确定的内容标记为需人工复审 历史参考案例： {references} 待审核内容： 类型: {content.content_type} 文本: {content.text} {'图片: 已上传' if content.image_base64 else ''} 请以 JSON 格式返回： {{"decision": "pass|block|review", "confidence": 0.0-1.0, "reason": "判断理由"}}""" try: response = await self._llm.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0, response_format={"type": "json_object"}, ) import json data = json.loads(response.choices[0].message.content or "{}") return AuditResult( decision=AuditDecision(data.get("decision", "review")), confidence=float(data.get("confidence", 0.5)), reason=data.get("reason", "LLM 判断"), references=[c.case_id for c in cases], ) except Exception as e: logger.error("LLM 审核失败: %s，降级为人工复审", e) return AuditResult( decision=AuditDecision.REVIEW, confidence=0.0, reason=f"LLM 异常: {str(e)[:50]}", ) async def main(): client = AsyncOpenAI(api_key="sk-xxx") # 假设 vector_store 已初始化 agent = MultimodalAuditAgent( llm_client=client, vector_store=None, # 替换为实际 FAISS/Milvus 客户端 top_k=5, ) # 测试正常内容 result = await agent.audit(ContentInput( text="这是一篇关于Python异步编程的技术文章", content_type="text", )) print(f"决策: {result.decision}, 置信度: {result.confidence}") # 测试对抗样本 result = await agent.audit(ContentInput( text="代`办`信`用`卡，需要的加V", content_type="text", )) print(f"决策: {result.decision}, 理由: {result.reason}") if __name__ == "__main__": asyncio.run(main())

代码中的关键设计：

• 分层拦截：硬规则（手机号、身份证）在第一步直接拦截，不浪费 LLM token。
• 检索降级：向量检索失败时返回空列表，审核降级为"纯 LLM 判断"，不会阻塞业务。
• LLM 失败兜底：LLM 调用异常时默认转人工复审，宁可多审不漏审。
• 白名单机制：历史 PASS 案例相似度 > 80% 直接放行，减少不必要的 LLM 调用。

四、边界分析与架构权衡

4.1 多模态的工程成本

文本审核和图片审核的延迟差异很大。文本 embedding + LLM 判断通常在 2-5 秒，图片加上 CLIP embedding 和多模态 LLM，延迟可能到 8-15 秒。对于实时评论这种场景，建议文本先审、图片异步补审——先让文本通过快速通道，图片结果回来后再做综合标记。

4.2 RAG 案例库的冷启动

系统上线初期没有足够的历史案例，RAG 的增强效果有限。建议从两方面解决：

• 种子案例：从已有的审核日志中抽取 500-1000 条高置信度案例人工标注后入库。
• 快速反馈：上线第一周，所有 REVIEW 结果强制走人工审核，审核结果实时入库，快速积累案例。

4.3 什么时候用 RAG 增强，什么时候纯 LLM？

4.4 审核一致性

RAG 引入了一个新问题：案例库的更新可能导致同一类型内容的审核结果前后不一致。一个月前被判通过的案例，因为案例库更新，类似内容可能被判为违规。

缓解方案：对每个审核结果记录检索到的 Top-K 案例 ID。当审核标准更新时，可以回溯"哪些历史审核决策可能需要重新评估"。

五、总结

多模态内容审核从"调 API"升级到"RAG 增强"，本质上是在做一件事：把审核标准的隐性知识变成可检索的显性向量。

纯 LLM 审核像一个刚入职的审核员——能力有，但不了解你的平台规范。RAG 相当于给他配了一本"审核判例手册"，每次审核前先翻翻类似案例是怎么判的，准确率自然上去了。

实施上建议分三步走：

1. 第一步：规则引擎兜底，手机号/身份证/银行卡直接拦截，不走 LLM。
2. 第二步：搭建案例向量库，从历史审核日志中抽取高置信度判例入库。
3. 第三步：接入 LLM 综合判断，RAG 案例作为上下文增强，人工复审兜底。

审核领域的容错原则是"宁可多审不漏审"。RAG 增强让这个原则有了工程上的落地路径。

下一篇预告：LangChain 自定义 LLM Wrapper，封装自部署模型为标准接口。