内容审核系统架构

内容审核是LLM治理的第一道防线，负责检测和过滤不当内容。

审核系统架构

审核层次结构

审核策略矩阵

审核内容分类

审核器接口设计

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from core.models import AuditRequest, AuditResult, AuditLevel, RiskLevel
class ContentAuditor(ABC):
"""内容审核器基类"""
def __init__(self, name: str, priority: int = 0):
self.name = name
self.priority = priority
@abstractmethod
async def audit(self, request: AuditRequest) -> AuditResult:
"""执行审核"""
pass
def get_score(self) -> int:
"""获取审核优先级分数（数字越小优先级越高）"""
return self.priority

多审核器链式执行

审核决策流程

class AuditDecision:
"""审核决策器"""
def __init__(self):
self.auditors: List[ContentAuditor] = []
self.block_on_first_fail = True  # 首次失败即阻断
async def evaluate(self, request: AuditRequest) -> AuditResult:
"""
评估审核请求
策略1: block_on_first_fail = True
- 任一审核器返回BLOCK，立即阻断
- 用于快速拦截高风险内容
策略2: block_on_first_fail = False
- 运行所有审核器
- 汇总所有审核结果
- 基于权重做最终决策
"""
results = []
blocked_reasons = []
# 按优先级排序审核器
sorted_auditors = sorted(self.auditors, key=lambda x: x.get_score())
for auditor in sorted_auditors:
result = await auditor.audit(request)
results.append(result)
if result.is_blocked:
blocked_reasons.extend(result.reasons)
if self.block_on_first_fail:
# 阻断并返回
return self._merge_blocked_results(
results,
blocked_reasons
)
# 所有审核器通过
return self._create_pass_result(results)
def _merge_blocked_results(
self,
results: List[AuditResult],
reasons: List[str]
) -> AuditResult:
"""合并阻断结果"""
return AuditResult(
request_id=results[0].request_id,
level=AuditLevel.BLOCK,
risk_level=max([r.risk_level for r in results]),
is_blocked=True,
reasons=reasons,
details={
"auditor_results": [r.dict() for r in results]
}
)
def _create_pass_result(
self,
results: List[AuditResult]
) -> AuditResult:
"""创建通过结果"""
return AuditResult(
request_id=results[0].request_id,
level=AuditLevel.PASS,
risk_level=RiskLevel.LOW,
is_blocked=False,
reasons=["所有审核器通过"],
details={
"auditor_results": [r.dict() for r in results]
}
)

审核性能优化

1. 缓存策略

from functools import lru_cache
from hashlib import md5
class CachedAuditor(ContentAuditor):
"""带缓存的审核器"""
def __init__(self, name: str, cache_size: int = 1000):
super().__init__(name)
self.cache_size = cache_size
def _get_cache_key(self, content: str) -> str:
"""生成缓存键"""
return md5(content.encode()).hexdigest()
@lru_cache(maxsize=1000)
def _audit_with_cache(self, content: str) -> AuditResult:
"""带缓存的审核"""
# 实际审核逻辑
pass

2. 异步并行审核

import asyncio
async def parallel_audit(
auditors: List[ContentAuditor],
request: AuditRequest
) -> List[AuditResult]:
"""并行执行审核"""
tasks = [auditor.audit(request) for auditor in auditors]
results = await asyncio.gather(*tasks)
return results

3. 短路优化

class ShortCircuitAuditor(ContentAuditor):
"""短路审核器（快速失败）"""
async def audit(self, request: AuditRequest) -> AuditResult:
# 先检查长度
if len(request.content) > 10000:
return self._block_result(request, "内容过长")
# 快速关键词检查
if self._quick_keyword_check(request.content):
return self._block_result(request, "包含敏感词")
# 完整审核
return await self._full_audit(request)

审核日志与追踪

from loguru import logger
from datetime import datetime
class AuditLogger:
"""审核日志记录器"""
def log_audit(
self,
request: AuditRequest,
result: AuditResult,
auditor_name: str
):
"""记录审核日志"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"request_id": request.request_id,
"auditor": auditor_name,
"content_length": len(request.content),
"result": result.level.value,
"blocked": result.is_blocked,
"risk_level": result.risk_level.value,
"reasons": result.reasons
}
if result.is_blocked:
logger.warning(f"🚫 审核阻断: {log_entry}")
else:
logger.debug(f"✅ 审核通过: {log_entry}")

测试框架

import pytest
from core.models import AuditRequest
@pytest.mark.asyncio
async def test_keyword_filter():
"""测试关键词过滤"""
from core.keyword_auditor import KeywordAuditor
auditor = KeywordAuditor()
# 测试正常内容
normal_request = AuditRequest(
request_id="test-001",
content="这是一条正常内容"
)
result = await auditor.audit(normal_request)
assert not result.is_blocked
# 测试违规内容
blocked_request = AuditRequest(
request_id="test-002",
content="这条内容包含暴力词汇"
)
result = await auditor.audit(blocked_request)
assert result.is_blocked
assert "暴力" in str(result.reasons)

学习要点

✅ 理解内容审核的多层架构 ✅ 掌握审核器链式执行模式 ✅ 了解审核决策策略 ✅ 学习性能优化方法（缓存、并行、短路） ✅ 实现审核日志追踪

下一步: 实现 关键词过滤器 🏷️