Skip to main content

内容审核

用例​

内容审核是在数字应用中维护安全、受尊重和高效环境的关键方面。本指南将讨论如何使用 Claude 对数字应用中的内容进行审核。

访问我们的内容审核指南以查看使用 Claude 进行内容审核的示例实现。

本指南主要关注应用程序内用户生成内容的审核。如果您需要有关审核与 Claude 交互的指导，请参阅我们的护栏指南。

使用 Claude 前的准备​

决定是否使用 Claude 进行内容审核​

以下是一些关键指标，表明您应该使用像 Claude 这样的 LLM 而不是传统的 ML 或基于规则的方法进行内容审核：

传统方法Claude 优势
传统 ML 方法需要大量工程资源、ML 专业知识和基础设施成本使用 Claude，您可以以更低的成本在更短的时间内建立复杂的审核系统
传统 ML 方法难以理解内容的语气、意图和上下文Claude 通过结合语义理解和快速审核决策能力弥补了这一差距
传统 ML 方法建立后，更改非常耗时且需要大量数据Claude 可以轻松适应审核政策的变化，无需大量重新标注数据
传统方法通常需要为每种支持的语言建立单独模型或翻译流程Claude 的多语言能力使其可以直接处理多种语言的内容

生成需要审核的内容示例​

在开发内容审核解决方案之前，首先创建应该标记和不应该标记的内容示例。确保包括可能难以处理的边缘案例和具有挑战性的场景。之后，检查示例以创建明确定义的审核类别列表。例如，社交媒体平台生成的示例可能包括：

allowed_user_comments = [
    '这部电影很棒，我真的很喜欢。主演真的太厉害了！',
    '我讨厌星期一。',
    '现在是投资黄金的好时机！'
]

disallowed_user_comments = [
    '立即删除这篇文章，否则你最好躲起来。我要找你和你的家人麻烦。',
    '远离5G手机！！他们正在用5G控制你。',
    '恭喜！您获得了1,000美元礼品卡。点击这里领取奖品！'
]

# 用于测试内容审核的示例用户评论
user_comments = allowed_user_comments + disallowed_user_comments

# 被认为不安全的内容类别列表
unsafe_categories = [
    '儿童剥削',
    '阴谋论',
    '仇恨',
    '大规模杀伤性武器',
    '知识产权',
    '非暴力犯罪',
    '隐私',
    '自残',
    '性犯罪',
    '性内容',
    '专业建议',
    '暴力犯罪'
]

选择合适的 Claude 模型​

在选择模型时，考虑数据量很重要。如果成本是一个考虑因素，像 Claude 3 Haiku 这样的较小模型是一个很好的选择，因为它具有成本效益。以下是对每月接收十亿帖子的社交媒体平台进行文本审核的成本估算：

内容规模及成本数量/金额
内容规模
每月帖子数10亿
每个帖子字符数100
总字符数1000亿
预估令牌数
输入令牌数286亿 (假设每3.5个字符1个令牌)
被标记消息百分比3%
每条被标记消息的输出令牌数50
总输出令牌数15亿
Claude 3 Haiku 预估成本
输入令牌成本2,860 MTok * $0.25/MTok = $715
输出令牌成本1,500 MTok * $1.25/MTok = $1,875
每月成本$715 + $1,875 = $2,590
Claude 3.5 Sonnet 预估成本
输入令牌成本2,860 MTok * $3.00/MTok = $8,580
输出令牌成本1,500 MTok * $15.00/MTok = $22,500
每月成本$8,580 + $22,500 = $31,080

构建有效的提示​

为了使用 Claude 进行内容审核，Claude 必须理解您应用程序的审核要求。让我们从编写一个定义审核需求的提示开始：

import anthropic
import json

# 初始化 Anthropic 客户端
client = anthropic.Anthropic()

def moderate_message(message, unsafe_categories):
    # 将不安全类别列表转换为字符串，每个类别占一行
    unsafe_category_str = '\n'.join(unsafe_categories)
    
    # 构建提示，包括消息和不安全类别
    assessment_prompt = f"""
    根据以下列出的不安全类别，判断该消息是否需要审核。

    消息：
    <message>{message}</message>

    不安全类别：
    <categories>
    {unsafe_category_str}
    </categories>

    仅使用以下 JSON 格式进行响应：
    {{
        "violation": <布尔字段，表示消息是否应该被审核>,
        "categories": [违反的类别列表，用逗号分隔],
        "explanation": [可选。仅在存在违规时包含。]
    }}"""

    # 向 Claude 发送内容审核请求
    response = client.messages.create(
        model="claude-3-haiku-20240307", # 使用 Haiku 模型以降低成本
        max_tokens=200,
        temperature=0, # 使用 0 温度以提高一致性
        messages=[
            {"role": "user", "content": assessment_prompt}
        ]
    )

    # 解析 Claude 的 JSON 响应
    assessment = json.loads(response.content[0].text)
    
    # 从评估中提取违规状态
    contains_violation = assessment['violation']
    
    # 如果存在违规，获取类别和解释；否则使用空默认值
    violated_categories = assessment.get('categories', []) if contains_violation else []
    explanation = assessment.get('explanation') if contains_violation else None

    return contains_violation, violated_categories, explanation

# 处理每条评论并打印结果
for comment in user_comments:
    print(f"\n评论: {comment}")
    violation, violated_categories, explanation = moderate_message(comment, unsafe_categories)
    if violation:
        print(f"违反类别: {', '.join(violated_categories)}")
        print(f"解释: {explanation}")
    else:
        print("未检测到问题。")

评估风险等级​

内容审核可以被视为多级分类问题，而不是简单的二元分类。以下是一个实现风险等级评估的示例：

import anthropic
import json

# 初始化 Anthropic 客户端
client = anthropic.Anthropic()

def assess_risk_level(message, unsafe_categories):
    # 将不安全类别列表转换为字符串
    unsafe_category_str = '\n'.join(unsafe_categories)
    
    # 构建包含消息、不安全类别和风险等级定义的提示
    assessment_prompt = f"""
    根据以下列出的不安全类别，评估该消息的风险等级。

    消息：
    <message>{message}</message>

    不安全类别：
    <categories>
    {unsafe_category_str}
    </categories>

    基于用户消息是否应该被审核的确信度分配风险等级：
    0 - 无风险
    1 - 低风险
    2 - 中等风险
    3 - 高风险

    仅使用以下 JSON 格式响应：
    {{
        "risk_level": <表示风险等级的数值>,
        "categories": [违反的类别列表，用逗号分隔],
        "explanation": <可选。仅在风险等级大于 0 时包含>
    }}"""

    # 向 Claude 发送风险评估请求
    response = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=200,
        temperature=0,
        messages=[
            {"role": "user", "content": assessment_prompt}
        ]
    )

    # 解析响应并提取相关信息
    assessment = json.loads(response.content[0].text)
    risk_level = assessment["risk_level"]
    violated_categories = assessment["categories"]
    explanation = assessment.get("explanation")

    return risk_level, violated_categories, explanation

部署建议​

在将解决方案部署到生产环境时，建议遵循以下最佳实践：

  1. 提供清晰的用户反馈：当用户输入被拦截或标记时，提供信息丰富的建设性反馈。
  2. 分析被审核的内容：跟踪被标记内容的类型，以识别趋势和潜在的改进领域。
  3. 持续评估和改进：定期评估内容审核系统的性能，使用精确度和召回率等指标。

批量处理​

对于需要处理大量内容的应用程序，建议使用批量处理来提高效率。以下是一个批量处理示例：

import anthropic
import json
from typing import List, Dict, Tuple

def batch_moderate_messages(messages: List[str], 
                          unsafe_categories: List[str], 
                          batch_size: int = 10) -> List[Dict]:
    # 将消息分成批次
    batches = [messages[i:i + batch_size] 
              for i in range(0, len(messages), batch_size)]
    results = []
    
    for batch in batches:
        # 构建批处理提示
        batch_prompt = "审核以下消息列表。对每条消息使用JSON格式返回结果：\n\n"
        for i, message in enumerate(batch):
            batch_prompt += f"消息 {i + 1}: {message}\n"
        
        # 添加类别信息
        batch_prompt += "\n不安全类别:\n"
        batch_prompt += "\n".join(unsafe_categories)
        
        response = client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=1000,
            temperature=0,
            messages=[{"role": "user", "content": batch_prompt}]
        )
        
        # 解析每个批次的结果
        batch_results = json.loads(response.content[0].text)
        results.extend(batch_results)
    
    return results

常见模式和最佳实践​

1. 多级审核​

实现多级审核系统可以平衡自动化和人工审查：

风险等级处理方式
低风险自动允许通过
中等风险进入审核队列，但允许发布
高风险自动阻止并要求人工审核

2. 上下文感知审核​

考虑消息的上下文来提高审核准确性：

def moderate_with_context(message: str, 
                         context: Dict,
                         unsafe_categories: List[str]) -> Dict:
    context_prompt = f"""
    考虑以下上下文评估消息:
    用户历史: {context.get('user_history', '无')}
    发布位置: {context.get('location', '无')}
    对话主题: {context.get('topic', '无')}
    
    消息:
    {message}
    
    不安全类别:
    {'\n'.join(unsafe_categories)}
    """
    
    # 实现审核逻辑
    return assessment

3. 本地化支持​

为不同地区调整审核标准：

def get_regional_categories(region: str) -> List[str]:
    regional_categories = {
        'EU': [...],  # 欧盟特定类别
        'US': [...],  # 美国特定类别
        'APAC': [...] # 亚太地区特定类别
    }
    return regional_categories.get(region, default_categories)

性能优化建议​

  1. 缓存策略

    • 缓存常见违规模式
    • 存储已审核内容的结果
    • 实现智能预缓存

  2. 批量处理优化

批量大小优点缺点
小批量 (1-10)低延迟，实时性好API调用频繁
中批量 (11-50)平衡性能和成本适中延迟
大批量 (50+)成本效益高较高延迟
  1. 负载均衡
    • 实现请求队列
    • 设置重试机制
    • 使用回退策略

监控和分析​

建立全面的监控系统：

  1. 关键指标

    • 审核延迟
    • 误报率
    • 漏报率
    • API 使用情况

  2. 日志记录

def log_moderation_event(message_id: str, 
                        result: Dict,
                        metadata: Dict):
    log_entry = {
        'timestamp': datetime.now().isoformat(),
        'message_id': message_id,
        'moderation_result': result,
        'processing_time': metadata.get('processing_time'),
        'model_version': metadata.get('model_version')
    }
    # 实现日志记录逻辑

错误处理和恢复​

实施稳健的错误处理机制对于生产环境至关重要：

class ModerationError(Exception):
    """自定义审核错误类"""
    pass

def safe_moderate_content(message: str, 
                         unsafe_categories: List[str],
                         max_retries: int = 3) -> Dict:
    for attempt in range(max_retries):
        try:
            result = moderate_message(message, unsafe_categories)
            return result
        except anthropic.APIError as e:
            if attempt == max_retries - 1:
                raise ModerationError(f"API错误：{str(e)}")
            time.sleep(2 ** attempt)  # 指数退避
        except json.JSONDecodeError:
            raise ModerationError("响应解析失败")
        except Exception as e:
            raise ModerationError(f"未知错误：{str(e)}")

可扩展性考虑​

为确保系统可以随着需求增长而扩展，考虑以下方面：

扩展维度建议措施
水平扩展使用多个服务实例
垂直扩展优化资源利用
地理扩展部署区域特定端点

API 限制管理​

class RateLimiter:
    def __init__(self, requests_per_minute: int):
        self.requests_per_minute = requests_per_minute
        self.requests = []
        self.lock = threading.Lock()

    def wait_if_needed(self):
        now = time.time()
        with self.lock:
            # 清理超过一分钟的旧请求
            self.requests = [req for req in self.requests 
                           if now - req < 60]
            
            if len(self.requests) >= self.requests_per_minute:
                sleep_time = 60 - (now - self.requests[0])
                time.sleep(max(0, sleep_time))
            
            self.requests.append(now)

审核策略管理​

创建灵活的审核策略系统：

class ModerationPolicy:
    def __init__(self):
        self.rules = []
        self.exceptions = []
        
    def add_rule(self, rule: Dict):
        """添加新的审核规则"""
        self.rules.append(rule)
        
    def add_exception(self, exception: Dict):
        """添加规则例外"""
        self.exceptions.append(exception)
        
    def evaluate(self, content: str) -> Tuple[bool, str]:
        """评估内容是否违反规则"""
        for exception in self.exceptions:
            if self._matches_criteria(content, exception):
                return False, "例外情况"
                
        for rule in self.rules:
            if self._matches_criteria(content, rule):
                return True, rule["reason"]
                
        return False, "通过"
        
    def _matches_criteria(self, content: str, criteria: Dict) -> bool:
        # 实现匹配逻辑
        pass

审计和合规​

建立审计跟踪系统：

class AuditLogger:
    def __init__(self, storage_client):
        self.storage_client = storage_client
        
    def log_moderation_decision(self, 
                              content_id: str,
                              decision: Dict,
                              metadata: Dict):
        audit_record = {
            "timestamp": datetime.utcnow().isoformat(),
            "content_id": content_id,
            "decision": decision,
            "metadata": metadata
        }
        
        self.storage_client.store(audit_record)

性能基准​

典型系统性能指标：

指标目标值
平均响应时间< 500ms
99百分位延迟< 2000ms
每秒处理请求数> 100
准确率> 95%

集成测试​

def test_moderation_system():
    test_cases = [
        {
            "content": "正常内容示例",
            "expected_result": {"violation": False}
        },
        {
            "content": "违规内容示例",
            "expected_result": {"violation": True}
        }
    ]
    
    for case in test_cases:
        result = moderate_message(case["content"], unsafe_categories)
        assert result["violation"] == case["expected_result"]["violation"], \
            f"测试失败：{case['content']}"

持续改进​

维护反馈循环以不断改进系统：

  1. 数据收集

    • 用户反馈
    • 审核决策
    • 系统性能指标

  2. 分析和优化

    • 定期审查规则
    • 更新类别
    • 优化提示

  3. 报告和指标

    • 生成性能报告
    • 跟踪关键指标
    • 识别改进机会

生产环境集成​

以下是将内容审核系统集成到生产环境的最佳实践：

队列系统集成​

import redis
from rq import Queue

class ModerationQueue:
    def __init__(self):
        self.redis_conn = redis.Redis()
        self.queue = Queue(connection=self.redis_conn)
        
    def enqueue_content(self, content: Dict):
        """将内容加入审核队列"""
        job = self.queue.enqueue(
            moderate_content,
            content,
            job_timeout='5m',
            result_ttl=86400  # 结果保留24小时
        )
        return job.id
        
    def get_result(self, job_id: str) -> Dict:
        """获取审核结果"""
        job = self.queue.fetch_job(job_id)
        if job is None:
            raise ValueError("任务不存在")
            
        if not job.is_finished:
            return {"status": "pending"}
            
        return {"status": "complete", "result": job.result}

数据库集成​

from sqlalchemy import create_engine, Column, Integer, String, JSON
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class ModerationResult(Base):
    __tablename__ = 'moderation_results'
    
    id = Column(Integer, primary_key=True)
    content_id = Column(String)
    result = Column(JSON)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    def to_dict(self):
        return {
            "id": self.id,
            "content_id": self.content_id,
            "result": self.result,
            "created_at": self.created_at.isoformat()
        }

高级功能实现​

1. 多语言支持​

def moderate_multilingual(content: str,
                         source_language: str,
                         unsafe_categories: List[str]) -> Dict:
    prompt = f"""
    以下是{source_language}语言的内容：
    {content}
    
    请使用原始语言评估此内容是否违反以下类别：
    {'\n'.join(unsafe_categories)}
    
    请用JSON格式返回结果。
    """
    
    # 实现审核逻辑
    return result

2. 图像文本提取和审核​

class MultimodalModerator:
    def __init__(self):
        self.text_moderator = TextModerator()
        self.image_analyzer = ImageAnalyzer()
        
    async def moderate_post(self, 
                          text: str, 
                          image_urls: List[str]) -> Dict:
        """审核包含文本和图像的帖子"""
        results = {
            "text_moderation": await self.text_moderator.moderate(text),
            "image_moderation": []
        }
        
        for url in image_urls:
            image_text = await self.image_analyzer.extract_text(url)
            if image_text:
                image_result = await self.text_moderator.moderate(image_text)
                results["image_moderation"].append({
                    "url": url,
                    "result": image_result
                })
                
        return results

性能优化技巧​

1. 缓存实现​

from functools import lru_cache
import hashlib

class ModerationCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 86400  # 24小时缓存
        
    def get_cache_key(self, content: str) -> str:
        """生成内容的缓存键"""
        return hashlib.md5(content.encode()).hexdigest()
        
    async def get_cached_result(self, content: str) -> Optional[Dict]:
        """获取缓存的审核结果"""
        cache_key = self.get_cache_key(content)
        result = await self.redis.get(cache_key)
        return json.loads(result) if result else None
        
    async def cache_result(self, content: str, result: Dict):
        """缓存审核结果"""
        cache_key = self.get_cache_key(content)
        await self.redis.setex(
            cache_key,
            self.ttl,
            json.dumps(result)
        )

2. 负载均衡器​

class LoadBalancer:
    def __init__(self, endpoints: List[str]):
        self.endpoints = endpoints
        self.current = 0
        self.lock = asyncio.Lock()
        
    async def get_next_endpoint(self) -> str:
        """获取下一个可用端点"""
        async with self.lock:
            endpoint = self.endpoints[self.current]
            self.current = (self.current + 1) % len(self.endpoints)
            return endpoint

监控和警报​

1. 指标收集​

class MetricsCollector:
    def __init__(self):
        self.metrics = defaultdict(Counter)
        
    def record_moderation(self, result: Dict):
        """记录审核结果指标"""
        self.metrics["total_moderations"].increment()
        if result["violation"]:
            self.metrics["violations"].increment()
            for category in result["categories"]:
                self.metrics[f"violation_{category}"].increment()
                
    def get_metrics(self) -> Dict:
        """获取当前指标"""
        return {k: v.value for k, v in self.metrics.items()}

2. 警报系统​

class AlertSystem:
    def __init__(self, thresholds: Dict):
        self.thresholds = thresholds
        self.notifier = NotificationService()
        
    async def check_alerts(self, metrics: Dict):
        """检查是否需要触发警报"""
        for metric, value in metrics.items():
            if metric in self.thresholds:
                threshold = self.thresholds[metric]
                if value > threshold:
                    await self.notifier.send_alert(
                        f"指标 {metric} 超过阈值: {value} > {threshold}"
                    )

API 接口设计​

1. RESTful API 实现​

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class ContentRequest(BaseModel):
    text: str
    metadata: dict = {}

class ModerationResponse(BaseModel):
    id: str
    status: str
    result: Optional[Dict] = None

@app.post("/moderate", response_model=ModerationResponse)
async def moderate_content(
    request: ContentRequest,
    background_tasks: BackgroundTasks
):
    """异步内容审核端点"""
    try:
        # 生成唯一任务ID
        task_id = str(uuid.uuid4())
        
        # 将审核任务加入后台队列
        background_tasks.add_task(
            process_moderation,
            task_id,
            request.text,
            request.metadata
        )
        
        return {
            "id": task_id,
            "status": "pending"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/moderate/{task_id}", response_model=ModerationResponse)
async def get_moderation_result(task_id: str):
    """获取审核结果端点"""
    result = await get_result_from_storage(task_id)
    if not result:
        raise HTTPException(status_code=404, detail="任务未找到")
    return result

2. WebSocket 实现实时审核​

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/moderate")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            content = await websocket.receive_text()
            result = await moderate_message(content, unsafe_categories)
            await websocket.send_json(result)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

高级配置管理​

1. 动态配置系统​

class ConfigManager:
    def __init__(self):
        self.config = {}
        self.observers = []
        
    async def load_config(self):
        """从配置源加载配置"""
        # 可以是数据库、文件或环境变量
        new_config = await load_from_source()
        self.update_config(new_config)
        
    def update_config(self, new_config: Dict):
        """更新配置并通知观察者"""
        self.config = new_config
        for observer in self.observers:
            observer(new_config)
            
    def register_observer(self, callback: Callable):
        """注册配置变更观察者"""
        self.observers.append(callback)

2. 规则引擎​

class Rule:
    def __init__(self, 
                 name: str,
                 condition: Callable,
                 action: Callable,
                 priority: int = 0):
        self.name = name
        self.condition = condition
        self.action = action
        self.priority = priority

class RuleEngine:
    def __init__(self):
        self.rules: List[Rule] = []
        
    def add_rule(self, rule: Rule):
        """添加新规则"""
        self.rules.append(rule)
        # 按优先级排序
        self.rules.sort(key=lambda x: x.priority, reverse=True)
        
    async def evaluate(self, content: Dict) -> List[Dict]:
        """评估内容并执行匹配的规则"""
        results = []
        for rule in self.rules:
            if await rule.condition(content):
                result = await rule.action(content)
                results.append({
                    "rule": rule.name,
                    "result": result
                })
        return results

测试框架​

1. 集成测试套件​

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_moderation_flow():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # 测试创建审核任务
        response = await client.post(
            "/moderate",
            json={
                "text": "测试内容",
                "metadata": {"source": "test"}
            }
        )
        assert response.status_code == 200
        task_id = response.json()["id"]
        
        # 测试获取结果
        result = await client.get(f"/moderate/{task_id}")
        assert result.status_code == 200

@pytest.mark.asyncio
async def test_websocket_connection():
    async with client.websocket_connect("/ws/moderate") as websocket:
        await websocket.send_text("测试内容")
        response = await websocket.receive_json()
        assert "result" in response

2. 性能测试工具​

import asyncio
import time
from statistics import mean, median

class PerformanceTester:
    def __init__(self, 
                 target_function: Callable,
                 sample_size: int = 100):
        self.target_function = target_function
        self.sample_size = sample_size
        self.results = []
        
    async def run_test(self):
        """执行性能测试"""
        start_time = time.time()
        
        tasks = [
            self.target_function()
            for _ in range(self.sample_size)
        ]
        
        results = await asyncio.gather(*tasks)
        self.results = results
        
        end_time = time.time()
        return self.calculate_metrics(end_time - start_time)
        
    def calculate_metrics(self, total_time: float) -> Dict:
        """计算性能指标"""
        return {
            "total_time": total_time,
            "average_time": total_time / self.sample_size,
            "success_rate": len([r for r in self.results if r]) / self.sample_size,
            "requests_per_second": self.sample_size / total_time
        }

部署指南​

1. Docker 配置​

# 基础镜像
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 复制依赖文件
COPY requirements.txt .

# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

2. Kubernetes 部署配置​

apiVersion: apps/v1
kind: Deployment
metadata:
  name: content-moderator
spec:
  replicas: 3
  selector:
    matchLabels:
      app: content-moderator
  template:
    metadata:
      labels:
        app: content-moderator
    spec:
      containers:
      - name: moderator
        image: content-moderator:latest
        resources:
          requests:
            memory: "256Mi"
            cpu: "200m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        env:
        - name: REDIS_HOST
          value: "redis-service"
        - name: DB_CONNECTION
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: connection-string

运维管理​

1. 监控仪表板配置​

from prometheus_client import Counter, Histogram, start_http_server

class MetricsServer:
    def __init__(self):
        # 定义指标
        self.request_count = Counter(
            'moderation_requests_total',
            '审核请求总数'
        )
        self.processing_time = Histogram(
            'moderation_processing_seconds',
            '审核处理时间',
            buckets=[0.1, 0.5, 1.0, 2.0, 5.0]
        )
        
    def start(self, port: int = 8000):
        """启动指标服务器"""
        start_http_server(port)
        
    def record_request(self):
        """记录新请求"""
        self.request_count.inc()
        
    @contextmanager
    def measure_time(self):
        """测量处理时间"""
        start_time = time.time()
        yield
        self.processing_time.observe(time.time() - start_time)

2. 日志管理​

import structlog
from typing import Any

class LogManager:
    def __init__(self):
        self.logger = structlog.get_logger()
        
    def setup_logging(self):
        """配置结构化日志"""
        structlog.configure(
            processors=[
                structlog.processors.TimeStamper(fmt="iso"),
                structlog.stdlib.add_log_level,
                structlog.processors.JSONRenderer()
            ],
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
        )
        
    def log_moderation(self,
                      content_id: str,
                      result: Dict,
                      extra: Dict[str, Any] = None):
        """记录审核事件"""
        log_data = {
            "content_id": content_id,
            "result": result
        }
        if extra:
            log_data.update(extra)
            
        self.logger.info("moderation_event", **log_data)

安全增强​

1. 加密服务​

from cryptography.fernet import Fernet
import base64

class EncryptionService:
    def __init__(self, key: bytes = None):
        self.key = key or Fernet.generate_key()
        self.cipher = Fernet(self.key)
        
    def encrypt_content(self, content: str) -> str:
        """加密内容"""
        encrypted = self.cipher.encrypt(content.encode())
        return base64.b64encode(encrypted).decode()
        
    def decrypt_content(self, encrypted_content: str) -> str:
        """解密内容"""
        decoded = base64.b64decode(encrypted_content)
        decrypted = self.cipher.decrypt(decoded)
        return decrypted.decode()

2. 访问控制​

from fastapi import Depends, Security
from fastapi.security import OAuth2PasswordBearer

class AccessControl:
    def __init__(self):
        self.oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
        self.permissions = {}
        
    async def verify_token(self, token: str = Depends(oauth2_scheme)):
        """验证访问令牌"""
        if not self._is_valid_token(token):
            raise HTTPException(
                status_code=401,
                detail="无效的访问令牌"
            )
        return token
        
    def require_permission(self, permission: str):
        """权限检查装饰器"""
        async def check_permission(
            token: str = Security(verify_token)
        ):
            if not self._has_permission(token, permission):
                raise HTTPException(
                    status_code=403,
                    detail="权限不足"
                )
            return True
        return Depends(check_permission)

高级应用场景​

1. 批量处理系统​

class BatchProcessor:
    def __init__(self, 
                 batch_size: int = 100,
                 max_wait_time: float = 1.0):
        self.batch_size = batch_size
        self.max_wait_time = max_wait_time
        self.batch = []
        self.last_process_time = time.time()
        
    async def add_item(self, item: Dict):
        """添加项目到批处理队列"""
        self.batch.append(item)
        
        if len(self.batch) >= self.batch_size or \
           time.time() - self.last_process_time > self.max_wait_time:
            await self.process_batch()
            
    async def process_batch(self):
        """处理当前批次"""
        if not self.batch:
            return
            
        try:
            results = await self._process_items(self.batch)
            await self._store_results(results)
        finally:
            self.batch = []
            self.last_process_time = time.time()

高级优化和扩展功能​

1. 内容分析器​

class ContentAnalyzer:
    def __init__(self):
        self.patterns = self._compile_patterns()
        self.tokenizer = Tokenizer()
        
    async def analyze(self, content: str) -> Dict:
        """深度分析内容特征"""
        return {
            "length": len(content),
            "token_count": len(self.tokenizer.tokenize(content)),
            "patterns": self._find_patterns(content),
            "language": await self._detect_language(content),
            "sentiment": await self._analyze_sentiment(content),
            "topics": await self._extract_topics(content)
        }
        
    async def _analyze_sentiment(self, content: str) -> Dict:
        """情感分析"""
        # 实现情感分析逻辑
        return {
            "positive": 0.0,
            "negative": 0.0,
            "neutral": 0.0
        }
        
    async def _extract_topics(self, content: str) -> List[str]:
        """主题提取"""
        topics = []
        # 实现主题提取逻辑
        return topics

2. 机器学习集成​

import numpy as np
from sklearn.ensemble import RandomForestClassifier

class MLModerator:
    def __init__(self):
        self.model = None
        self.vectorizer = None
        self.label_encoder = None
        
    async def train(self, 
                   texts: List[str],
                   labels: List[str]):
        """训练模型"""
        # 特征提取
        X = self.vectorizer.fit_transform(texts)
        y = self.label_encoder.fit_transform(labels)
        
        # 训练模型
        self.model = RandomForestClassifier(n_estimators=100)
        self.model.fit(X, y)
        
    async def predict(self, text: str) -> Dict:
        """预测内容分类"""
        features = self.vectorizer.transform([text])
        prediction = self.model.predict_proba(features)[0]
        
        return {
            "class": self.label_encoder.inverse_transform(
                [prediction.argmax()]
            )[0],
            "confidence": float(prediction.max())
        }

可扩展性优化​

1. 分布式处理框架​

from distributed import Client, LocalCluster

class DistributedModerator:
    def __init__(self, scheduler_address: str = None):
        if scheduler_address:
            self.client = Client(scheduler_address)
        else:
            self.cluster = LocalCluster()
            self.client = Client(self.cluster)
            
    async def process_batch(self, 
                          items: List[Dict],
                          batch_size: int = 1000):
        """分布式处理大批量内容"""
        # 划分批次
        batches = [
            items[i:i + batch_size]
            for i in range(0, len(items), batch_size)
        ]
        
        # 分发任务
        futures = []
        for batch in batches:
            future = self.client.submit(
                moderate_batch,
                batch
            )
            futures.append(future)
            
        # 收集结果
        results = await self.client.gather(futures)
        return [item for batch in results for item in batch]

2. 缓存优化​

class CacheOptimizer:
    def __init__(self,
                 redis_client,
                 max_cache_size: int = 10000):
        self.redis = redis_client
        self.max_cache_size = max_cache_size
        self.cache_stats = defaultdict(int)
        
    async def optimize_cache(self):
        """优化缓存策略"""
        # 分析缓存命中率
        total_requests = sum(self.cache_stats.values())
        hit_rate = self.cache_stats["hits"] / total_requests
        
        # 根据命中率调整缓存策略
        if hit_rate < 0.5:
            await self._adjust_cache_ttl()
            
        # 清理低使用率的缓存项
        await self._cleanup_unused_cache()
        
    async def _adjust_cache_ttl(self):
        """调整缓存过期时间"""
        keys = await self.redis.keys("*")
        for key in keys:
            usage_count = self.cache_stats[key]
            if usage_count > 100:
                # 增加高使用率项的 TTL
                await self.redis.expire(key, 86400 * 7)  # 7天
            elif usage_count < 10:
                # 减少低使用率项的 TTL
                await self.redis.expire(key, 3600)  # 1小时

内容过滤增强​

1. 高级过滤规则引擎​

class FilterRule:
    def __init__(self,
                 name: str,
                 patterns: List[str],
                 score: float,
                 category: str):
        self.name = name
        self.patterns = [re.compile(p) for p in patterns]
        self.score = score
        self.category = category
        
class AdvancedFilter:
    def __init__(self):
        self.rules = []
        self.threshold = 0.7
        
    def add_rule(self, rule: FilterRule):
        """添加过滤规则"""
        self.rules.append(rule)
        
    async def evaluate(self, content: str) -> Dict:
        """评估内容"""
        results = []
        total_score = 0
        
        for rule in self.rules:
            matches = []
            for pattern in rule.patterns:
                if pattern.search(content):
                    matches.append(pattern.pattern)
                    
            if matches:
                score = rule.score * len(matches)
                total_score += score
                results.append({
                    "rule": rule.name,
                    "category": rule.category,
                    "matches": matches,
                    "score": score
                })
                
        return {
            "results": results,
            "total_score": total_score,
            "is_violation": total_score > self.threshold
        }

实时处理优化​

1. 流处理系统​

from asyncio import Queue
from typing import AsyncGenerator

class StreamProcessor:
    def __init__(self, buffer_size: int = 1000):
        self.queue = Queue(maxsize=buffer_size)
        self.processors = []
        
    async def add_processor(self, processor: Callable):
        """添加流处理器"""
        self.processors.append(processor)
        
    async def process_stream(self, 
                           stream: AsyncGenerator) -> AsyncGenerator:
        """处理数据流"""
        async for item in stream:
            # 依次应用所有处理器
            processed_item = item
            for processor in self.processors:
                processed_item = await processor(processed_item)
                if processed_item is None:
                    break
                    
            if processed_item is not None:
                await self.queue.put(processed_item)
                
            # 输出处理后的数据
            while not self.queue.empty():
                yield await self.queue.get()

2. 实时统计分析​

class RealTimeAnalytics:
    def __init__(self, window_size: int = 3600):
        self.window_size = window_size  # 统计窗口大小（秒）
        self.stats = defaultdict(lambda: deque(maxlen=window_size))
        
    async def record_event(self, 
                          event_type: str,
                          value: float = 1.0):
        """记录事件"""
        current_time = int(time.time())
        self.stats[event_type].append((current_time, value))
        
    async def get_statistics(self, 
                           event_type: str) -> Dict:
        """获取实时统计数据"""
        current_time = int(time.time())
        window_start = current_time - self.window_size
        
        # 过滤窗口内的数据
        window_data = [
            value for timestamp, value in self.stats[event_type]
            if timestamp > window_start
        ]
        
        if not window_data:
            return {
                "count": 0,
                "sum": 0,
                "avg": 0,
                "min": 0,
                "max": 0
            }
            
        return {
            "count": len(window_data),
            "sum": sum(window_data),
            "avg": sum(window_data) / len(window_data),
            "min": min(window_data),
            "max": max(window_data)
        }

高级报告生成​

1. 报告生成器​

class ReportGenerator:
    def __init__(self):
        self.template_engine = Jinja2Templates(directory="templates")
        
    async def generate_report(self, 
                            data: Dict,
                            template: str = "default_report.html") -> str:
        """生成报告"""
        context = {
            "timestamp": datetime.now().isoformat(),
            "data": data,
            "summary": await self._generate_summary(data),
            "charts": await self._generate_charts(data),
            "recommendations": await self._generate_recommendations(data)
        }
        
        return self.template_engine.render(template, context)
        
    async def _generate_summary(self, data: Dict) -> Dict:
        """生成数据摘要"""
        return {
            "total_items": len(data),
            "violation_rate": sum(1 for item in data if item["is_violation"]) / len(data),
            "top_categories": Counter(
                item["category"] for item in data
            ).most_common(5)
        }
        
    async def _generate_recommendations(self, data: Dict) -> List[str]:
        """生成改进建议"""
        recommendations = []
        stats = await self._analyze_patterns(data)
        
        if stats["violation_rate"] > 0.1:
            recommendations.append(
                "建议加强内容审核标准，当前违规率较高"
            )
            
        return recommendations

2. 可视化生成器​

class VisualizationGenerator:
    def __init__(self):
        self.chart_configs = self._load_chart_configs()
        
    async def generate_chart(self,
                           data: List[Dict],
                           chart_type: str) -> Dict:
        """生成图表配置"""
        if chart_type == "trend":
            return await self._generate_trend_chart(data)
        elif chart_type == "distribution":
            return await self._generate_distribution_chart(data)
        else:
            raise ValueError(f"不支持的图表类型: {chart_type}")
            
    async def _generate_trend_chart(self, data: List[Dict]) -> Dict:
        """生成趋势图配置"""
        # 按时间分组数据
        grouped_data = defaultdict(list)
        for item in data:
            timestamp = item["timestamp"].date()
            grouped_data[timestamp].append(item)
            
        return {
            "type": "line",
            "data": {
                "labels": sorted(grouped_data.keys()),
                "datasets": [{
                    "label": "违规数量",
                    "data": [
                        len([i for i in group if i["is_violation"]])
                        for group in grouped_data.values()
                    ]
                }]
            },
            "options": self.chart_configs["trend"]
        }

高级优化特性​

1. 性能优化管理器​

class PerformanceManager:
    def __init__(self):
        self.metrics = {}
        self.thresholds = {
            "response_time": 500,  # 毫秒
            "memory_usage": 0.8,   # 80% 内存使用率
            "cpu_usage": 0.7       # 70% CPU 使用率
        }
        
    async def monitor_performance(self):
        """监控系统性能"""
        while True:
            metrics = {
                "response_time": await self._get_avg_response_time(),
                "memory_usage": await self._get_memory_usage(),
                "cpu_usage": await self._get_cpu_usage()
            }
            
            await self._analyze_metrics(metrics)
            await self._optimize_if_needed(metrics)
            
            await asyncio.sleep(60)  # 每分钟检查一次
            
    async def _optimize_if_needed(self, metrics: Dict):
        """根据需要进行优化"""
        if metrics["memory_usage"] > self.thresholds["memory_usage"]:
            await self._optimize_memory()
            
        if metrics["cpu_usage"] > self.thresholds["cpu_usage"]:
            await self._optimize_cpu()
            
    async def _optimize_memory(self):
        """内存优化"""
        gc.collect()  # 触发垃圾回收
        # 其他内存优化策略

2. 自适应配置管理器​

class AdaptiveConfig:
    def __init__(self):
        self.config = self._load_default_config()
        self.history = []
        
    async def adjust_config(self, metrics: Dict):
        """根据性能指标调整配置"""
        self.history.append(metrics)
        
        if len(self.history) >= 10:  # 至少收集10个数据点
            # 分析趋势
            trend = self._analyze_trend()
            
            # 根据趋势调整配置
            if trend["response_time"] > 0:  # 响应时间在增加
                await self._optimize_for_speed()
            
            if trend["error_rate"] > 0:  # 错误率在增加
                await self._optimize_for_stability()
                
            # 清理旧数据
            if len(self.history) > 100:
                self.history = self.history[-100:]
                
    async def _optimize_for_speed(self):
        """优化性能配置"""
        self.config.update({
            "cache_size": self.config["cache_size"] * 1.2,
            "batch_size": max(10, self.config["batch_size"] - 10),
            "parallel_workers": min(
                cpu_count() * 2,
                self.config["parallel_workers"] + 1
            )
        })

高级调试和诊断​

1. 诊断工具​

class DiagnosticTool:
    def __init__(self):
        self.traces = []
        self.error_patterns = self._load_error_patterns()
        
    async def analyze_system(self) -> Dict:
        """系统诊断分析"""
        return {
            "performance": await self._analyze_performance(),
            "errors": await self._analyze_errors(),
            "resources": await self._analyze_resources(),
            "recommendations": await self._generate_recommendations()
        }
        
    async def _analyze_errors(self) -> Dict:
        """错误模式分析"""
        error_counts = Counter()
        error_impacts = defaultdict(list)
        
        for trace in self.traces:
            if "error" in trace:
                error_type = self._classify_error(trace["error"])
                error_counts[error_type] += 1
                error_impacts[error_type].append(
                    self._calculate_impact(trace)
                )
                
        return {
            "frequencies": dict(error_counts),
            "impacts": {
                error_type: statistics.mean(impacts)
                for error_type, impacts in error_impacts.items()
            }
        }

2. 系统健康检查器​

class HealthChecker:
    def __init__(self):
        self.checks = []
        self.status_history = deque(maxlen=1000)
        
    async def add_check(self, 
                       name: str,
                       check_func: Callable,
                       interval: int = 60):
        """添加健康检查项"""
        self.checks.append({
            "name": name,
            "func": check_func,
            "interval": interval,
            "last_run": 0,
            "status": "unknown"
        })
        
    async def run_checks(self):
        """执行所有检查"""
        results = {}
        current_time = time.time()
        
        for check in self.checks:
            if current_time - check["last_run"] >= check["interval"]:
                try:
                    status = await check["func"]()
                    check["status"] = status
                    check["last_run"] = current_time
                except Exception as e:
                    check["status"] = "error"
                    logger.error(f"Health check failed: {str(e)}")
                    
                results[check["name"]] = check["status"]
                
        self.status_history.append({
            "timestamp": current_time,
            "results": results
        })
        
        return results

3. 状态监控器​

class StatusMonitor:
    def __init__(self):
        self.components = {}
        self.alerts = []
        
    async def register_component(self,
                               name: str,
                               check_func: Callable):
        """注册组件监控"""
        self.components[name] = {
            "check_func": check_func,
            "status": "unknown",
            "last_check": None,
            "failures": 0
        }
        
    async def monitor(self):
        """监控所有组件"""
        while True:
            for name, component in self.components.items():
                try:
                    status = await component["check_func"]()
                    await self._update_status(name, status)
                except Exception as e:
                    await self._handle_failure(name, e)
                    
            await asyncio.sleep(30)  # 30秒检查间隔
            
    async def _handle_failure(self, 
                            component_name: str,
                            error: Exception):
        """处理组件故障"""
        component = self.components[component_name]
        component["failures"] += 1
        
        if component["failures"] >= 3:  # 连续3次失败
            await self._trigger_alert({
                "component": component_name,
                "error": str(error),
                "timestamp": datetime.now().isoformat()
            })

异常处理和恢复机制​

1. 高级异常处理器​

class ExceptionHandler:
    def __init__(self):
        self.recovery_strategies = {}
        self.error_counts = defaultdict(int)
        self.last_errors = defaultdict(list)
        
    async def handle_exception(self,
                             error: Exception,
                             context: Dict) -> bool:
        """处理异常"""
        error_type = type(error).__name__
        self.error_counts[error_type] += 1
        
        # 记录错误详情
        self.last_errors[error_type].append({
            "timestamp": datetime.now().isoformat(),
            "message": str(error),
            "context": context
        })
        
        # 尝试恢复
        if error_type in self.recovery_strategies:
            try:
                await self.recovery_strategies[error_type](error, context)
                return True
            except Exception as recovery_error:
                logger.error(
                    f"Recovery failed: {str(recovery_error)}"
                )
                return False
        
        return False
        
    async def register_recovery_strategy(self,
                                       error_type: str,
                                       strategy: Callable):
        """注册错误恢复策略"""
        self.recovery_strategies[error_type] = strategy

2. 自动恢复系统​

class AutoRecoverySystem:
    def __init__(self):
        self.backup_manager = BackupManager()
        self.state_manager = StateManager()
        
    async def attempt_recovery(self,
                             error: Exception,
                             max_attempts: int = 3) -> bool:
        """尝试自动恢复"""
        for attempt in range(max_attempts):
            try:
                # 保存当前状态
                await self.state_manager.save_state()
                
                # 执行恢复步骤
                await self._execute_recovery_steps(error)
                
                # 验证恢复结果
                if await self._verify_recovery():
                    return True
                    
            except Exception as recovery_error:
                logger.error(
                    f"Recovery attempt {attempt + 1} failed: {str(recovery_error)}"
                )
                
                # 回滚到上一个状态
                await self.state_manager.rollback()
                
        return False
        
    async def _execute_recovery_steps(self, error: Exception):
        """执行恢复步骤"""
        # 根据错误类型确定恢复步骤
        steps = self._get_recovery_steps(error)
        
        for step in steps:
            await step()
            
    async def _verify_recovery(self) -> bool:
        """验证恢复结果"""
        # 执行系统健康检查
        health_check = await self._check_system_health()
        
        # 验证关键功能
        functionality_check = await self._verify_functionality()
        
        return health_check and functionality_check

高级安全特性​

1. 安全管理器​

class SecurityManager:
    def __init__(self):
        self.auth_provider = AuthenticationProvider()
        self.encryption_service = EncryptionService()
        self.audit_logger = AuditLogger()
        
    async def secure_operation(self,
                             operation: Callable,
                             context: Dict) -> Any:
        """安全操作包装器"""
        # 验证权限
        if not await self._check_permissions(context):
            raise PermissionError("Unauthorized operation")
            
        # 加密敏感数据
        secured_context = await self._encrypt_sensitive_data(context)
        
        try:
            # 执行操作
            result = await operation(**secured_context)
            
            # 记录审计日志
            await self.audit_logger.log_operation(
                operation.__name__,
                context,
                success=True
            )
            
            return result
            
        except Exception as e:
            # 记录失败
            await self.audit_logger.log_operation(
                operation.__name__,
                context,
                success=False,
                error=str(e)
            )
            raise

2. 威胁检测系统​

class ThreatDetector:
    def __init__(self):
        self.rules = self._load_threat_rules()
        self.detection_history = []
        
    async def analyze_request(self,
                            request_data: Dict) -> Dict:
        """分析请求中的潜在威胁"""
        threats = []
        risk_score = 0
        
        # 应用检测规则
        for rule in self.rules:
            if await self._apply_rule(rule, request_data):
                threats.append(rule["name"])
                risk_score += rule["risk_weight"]
                
        result = {
            "threats_detected": threats,
            "risk_score": risk_score,
            "timestamp": datetime.now().isoformat(),
            "is_dangerous": risk_score > 75
        }
        
        # 记录检测历史
        self.detection_history.append(result)
        
        return result
        
    async def _apply_rule(self,
                         rule: Dict,
                         data: Dict) -> bool:
        """应用检测规则"""
        try:
            pattern = rule["pattern"]
            if isinstance(pattern, str):
                return re.search(pattern, str(data)) is not None
            elif callable(pattern):
                return await pattern(data)
        except Exception as e:
            logger.error(f"Rule application failed: {str(e)}")
            return False

高级数据处理和分析​

1. 数据预处理器​

class DataPreprocessor:
    def __init__(self):
        self.transformers = {}
        self.validators = {}
        
    async def process_data(self, 
                          data: Dict,
                          schema: str) -> Dict:
        """数据预处理"""
        # 数据验证
        if not await self._validate_data(data, schema):
            raise ValueError("数据验证失败")
            
        processed_data = data.copy()
        
        # 应用转换器
        for field, transformer in self.transformers.items():
            if field in processed_data:
                processed_data[field] = await transformer(
                    processed_data[field]
                )
                
        return processed_data
        
    async def add_transformer(self,
                            field: str,
                            transformer: Callable):
        """添加数据转换器"""
        self.transformers[field] = transformer
        
    async def add_validator(self,
                          schema: str,
                          validator: Callable):
        """添加数据验证器"""
        self.validators[schema] = validator

2. 高级分析引擎​

class AnalyticsEngine:
    def __init__(self):
        self.models = {}
        self.processors = []
        
    async def analyze_data(self,
                          data: List[Dict],
                          analysis_type: str) -> Dict:
        """执行数据分析"""
        # 数据预处理
        processed_data = await self._preprocess_data(data)
        
        # 特征提取
        features = await self._extract_features(processed_data)
        
        # 应用分析模型
        if analysis_type not in self.models:
            raise ValueError(f"未知的分析类型: {analysis_type}")
            
        results = await self.models[analysis_type].analyze(features)
        
        # 后处理结果
        return await self._postprocess_results(results)
        
    async def _extract_features(self,
                              data: List[Dict]) -> np.ndarray:
        """特征提取"""
        features = []
        for item in data:
            item_features = []
            for processor in self.processors:
                processed = await processor(item)
                item_features.extend(processed)
            features.append(item_features)
            
        return np.array(features)

系统优化与扩展​

1. 动态资源分配器​

class ResourceAllocator:
    def __init__(self):
        self.resources = {}
        self.allocation_strategy = "balanced"
        
    async def allocate_resources(self,
                               task: Dict) -> Dict:
        """分配资源"""
        required_resources = await self._calculate_requirements(task)
        available_resources = await self._get_available_resources()
        
        allocation = {}
        
        # 根据策略分配资源
        if self.allocation_strategy == "balanced":
            allocation = await self._balanced_allocation(
                required_resources,
                available_resources
            )
        elif self.allocation_strategy == "priority":
            allocation = await self._priority_based_allocation(
                required_resources,
                available_resources,
                task.get("priority", 0)
            )
            
        # 更新资源状态
        await self._update_resource_status(allocation)
        
        return allocation
        
    async def _balanced_allocation(self,
                                 required: Dict,
                                 available: Dict) -> Dict:
        """均衡分配策略"""
        allocation = {}
        for resource_type, amount in required.items():
            if resource_type in available:
                allocation[resource_type] = min(
                    amount,
                    available[resource_type]
                )
        return allocation

2. 系统扩展管理器​

class ExtensionManager:
    def __init__(self):
        self.extensions = {}
        self.dependencies = defaultdict(list)
        
    async def load_extension(self,
                           name: str,
                           config: Dict) -> bool:
        """加载系统扩展"""
        # 检查依赖
        if not await self._check_dependencies(name):
            raise DependencyError(f"扩展 {name} 的依赖检查失败")
            
        try:
            # 初始化扩展
            extension = await self._initialize_extension(name, config)
            
            # 注册扩展
            self.extensions[name] = {
                "instance": extension,
                "config": config,
                "status": "active"
            }
            
            # 运行扩展初始化钩子
            await extension.on_load()
            
            return True
            
        except Exception as e:
            logger.error(f"扩展 {name} 加载失败: {str(e)}")
            return False
            
    async def _check_dependencies(self, name: str) -> bool:
        """检查扩展依赖"""
        if name not in self.dependencies:
            return True
            
        for dep in self.dependencies[name]:
            if dep not in self.extensions:
                return False
            if self.extensions[dep]["status"] != "active":
                return False
                
        return True