内容审核 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 多模态审核流水线架构这个流水线的关键设计点多模态 pipeline 解耦文本、图片、视频分别走各自的 embedding 管道最后在向量检索层统一。这样新增模态不需要改核心审核逻辑。规则库 向量检索双路向量检索负责找相似案例规则引擎负责硬匹配。比如手机号正则检测不需要过 LLM直接用规则拦截。反馈闭环人工复审的结果会写回向量库形成审核案例的持续积累。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(ge0.0, le1.0) reason: str references: list[str] Field(default_factorylist) class MultimodalAuditAgent: 多模态内容审核 AgentRAG 增强版。 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, r1[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), timeouttimeout ) except asyncio.TimeoutError: logger.error(审核超时降级为人工复审) return AuditResult( decisionAuditDecision.REVIEW, confidence0.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( decisionAuditDecision.BLOCK, confidence1.0, reasonhard_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( decisionAuditDecision.PASS, confidence0.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( modelclip-vit-large-patch14, inputcontent.text or image content, ) except Exception as e: logger.warning(图片 embedding 失败: %s降级为文本, e) response await self._llm.embeddings.create( modelself._embedding_model, inputcontent.text or unknown image, ) else: response await self._llm.embeddings.create( modelself._embedding_model, inputcontent.text, ) emb np.array(response.data[0].embedding, dtypenp.float32) return emb async def _retrieve_cases( self, query_emb: np.ndarray ) - list[AuditCase]: 向量检索相似历史审核案例。 try: results self._vector_store.search( query_emb.tolist(), top_kself._top_k ) cases [] for r in results: if r.get(score, 0) self._threshold: cases.append(AuditCase( case_idr[case_id], content_textr.get(text, ), content_typer.get(type, text), decisionAuditDecision(r.get(decision, review)), reasonr.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( modelgpt-4o, messages[{role: user, content: prompt}], temperature0, response_format{type: json_object}, ) import json data json.loads(response.choices[0].message.content or {}) return AuditResult( decisionAuditDecision(data.get(decision, review)), confidencefloat(data.get(confidence, 0.5)), reasondata.get(reason, LLM 判断), references[c.case_id for c in cases], ) except Exception as e: logger.error(LLM 审核失败: %s降级为人工复审, e) return AuditResult( decisionAuditDecision.REVIEW, confidence0.0, reasonfLLM 异常: {str(e)[:50]}, ) async def main(): client AsyncOpenAI(api_keysk-xxx) # 假设 vector_store 已初始化 agent MultimodalAuditAgent( llm_clientclient, vector_storeNone, # 替换为实际 FAISS/Milvus 客户端 top_k5, ) # 测试正常内容 result await agent.audit(ContentInput( text这是一篇关于Python异步编程的技术文章, content_typetext, )) print(f决策: {result.decision}, 置信度: {result.confidence}) # 测试对抗样本 result await agent.audit(ContentInput( text代办信用卡需要的加V, content_typetext, )) 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场景推荐方案通用内容审核无行业特殊规范纯 LLM 规则引擎足够垂直行业医疗、金融、法律必须 RAG 增强行业规范太特殊高对抗场景水军、绕过检测RAG 对抗样本库持续更新黑名单多语言内容每种语言独立案例库不能混用视频内容关键帧提取 字幕 OCR 文本审核组合4.4 审核一致性RAG 引入了一个新问题案例库的更新可能导致同一类型内容的审核结果前后不一致。一个月前被判通过的案例因为案例库更新类似内容可能被判为违规。缓解方案对每个审核结果记录检索到的 Top-K 案例 ID。当审核标准更新时可以回溯哪些历史审核决策可能需要重新评估。五、总结多模态内容审核从调 API升级到RAG 增强本质上是在做一件事把审核标准的隐性知识变成可检索的显性向量。纯 LLM 审核像一个刚入职的审核员——能力有但不了解你的平台规范。RAG 相当于给他配了一本审核判例手册每次审核前先翻翻类似案例是怎么判的准确率自然上去了。实施上建议分三步走第一步规则引擎兜底手机号/身份证/银行卡直接拦截不走 LLM。第二步搭建案例向量库从历史审核日志中抽取高置信度判例入库。第三步接入 LLM 综合判断RAG 案例作为上下文增强人工复审兜底。审核领域的容错原则是宁可多审不漏审。RAG 增强让这个原则有了工程上的落地路径。下一篇预告LangChain 自定义 LLM Wrapper封装自部署模型为标准接口。