引言

随着AIGC技术的快速发展，AI生成内容的安全性和合规性成为重要课题。内容审核需要对文本、图像、视频等多模态数据进行智能检测，识别违规、有害或不当内容。华为CANN平台通过高效的多模态模型推理和并行处理能力，为大规模内容审核提供了强大的技术支撑。

本文将介绍如何使用CANN构建多模态内容审核系统，实现高效、准确的内容安全检测。

相关链接：

• CANN组织
• ops-nn仓库

一、CANN文本内容审核

1.1 文本分类模型部署

使用CANN部署文本内容审核模型：

import torch
import torch_npu
from transformers import AutoTokenizer, AutoModelForSequenceClassification

class CANNTextModerator:
    """CANN文本内容审核器"""
    
    def __init__(self, model_name="bert-base-uncased"):
        self.device = "npu:0"
        print(f"正在加载文本审核模型到 {self.device}...")
        
        # 加载模型
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name,
            num_labels=5,  # 正常、暴力、色情、政治敏感、其他违规
            torch_dtype=torch.float16
        ).to(self.device)
        
        self.model.eval()
        self._enable_cann_optimizations()
        
        # 类别标签
        self.labels = ['normal', 'violence', 'nsfw', 'political', 'other_violation']
        
        print("文本审核模型加载完成")
    
    def _enable_cann_optimizations(self):
        """启用CANN优化"""
        for param in self.model.parameters():
            param.requires_grad = False
        
        torch_npu.npu.set_option({
            'ACL_OP_COMPILER_CACHE_MODE': '1'
        })
    
    def moderate_text(self, text, threshold=0.7):
        """审核单条文本"""
        # 编码输入
        inputs = self.tokenizer(
            text,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        ).to(self.device)
        
        with torch.no_grad():
            # CANN加速推理
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1)
        
        # 获取预测结果
        max_prob, pred_label = torch.max(probs, dim=1)
        max_prob = max_prob.item()
        pred_label = pred_label.item()
        
        # 判断是否违规
        is_safe = (self.labels[pred_label] == 'normal' and max_prob > threshold)
        
        result = {
            'text': text,
            'is_safe': is_safe,
            'category': self.labels[pred_label],
            'confidence': max_prob,
            'all_scores': {
                label: prob.item() 
                for label, prob in zip(self.labels, probs[0])
            }
        }
        
        return result
    
    def batch_moderate(self, texts, batch_size=32):
        """批量审核文本"""
        results = []
        
        for i in range(0, len(texts), batch_size):
            batch_texts = texts[i:i+batch_size]
            
            # 批量编码
            inputs = self.tokenizer(
                batch_texts,
                return_tensors="pt",
                truncation=True,
                max_length=512,
                padding=True
            ).to(self.device)
            
            with torch.no_grad():
                outputs = self.model(**inputs)
                probs = torch.softmax(outputs.logits, dim=-1)
            
            # 处理每条结果
            for text, prob_dist in zip(batch_texts, probs):
                max_prob, pred_label = torch.max(prob_dist, dim=0)
                
                result = {
                    'text': text,
                    'is_safe': (self.labels[pred_label.item()] == 'normal'),
                    'category': self.labels[pred_label.item()],
                    'confidence': max_prob.item()
                }
                results.append(result)
        
        return results

# 使用示例
if __name__ == "__main__":
    # 初始化审核器
    moderator = CANNTextModerator()
    
    # 测试文本
    test_texts = [
        "This is a normal conversation about technology.",
        "Let's discuss the latest AI developments.",
        "I love programming and building cool projects."
    ]
    
    # 批量审核
    results = moderator.batch_moderate(test_texts)
    
    print("审核结果:")
    for result in results:
        status = "✓ 安全" if result['is_safe'] else "✗ 违规"
        print(f"{status} | {result['category']} ({result['confidence']:.2%})")
        print(f"  文本: {result['text'][:50]}...")
        print()

1.2 敏感词检测

实现高效的敏感词过滤：

class CANNSensitiveWordFilter:
    """CANN敏感词过滤器"""
    
    def __init__(self):
        self.sensitive_words = set()
        self.load_sensitive_words()
    
    def load_sensitive_words(self):
        """加载敏感词库"""
        # 示例敏感词（实际应用中从文件加载）
        self.sensitive_words = {
            'badword1', 'badword2', 'badword3',
            # 更多敏感词...
        }
        print(f"已加载 {len(self.sensitive_words)} 个敏感词")
    
    def contains_sensitive_words(self, text):
        """检查是否包含敏感词"""
        text_lower = text.lower()
        found_words = []
        
        for word in self.sensitive_words:
            if word in text_lower:
                found_words.append(word)
        
        return len(found_words) > 0, found_words
    
    def mask_sensitive_words(self, text, mask_char='*'):
        """屏蔽敏感词"""
        masked_text = text
        
        for word in self.sensitive_words:
            if word in masked_text.lower():
                # 替换为星号
                masked_text = masked_text.replace(
                    word,
                    mask_char * len(word)
                )
        
        return masked_text

# 使用示例
word_filter = CANNSensitiveWordFilter()

text = "This is a test message with some content."
has_sensitive, found = word_filter.contains_sensitive_words(text)

if has_sensitive:
    print(f"发现敏感词: {found}")
    masked = word_filter.mask_sensitive_words(text)
    print(f"屏蔽后: {masked}")
else:
    print("未发现敏感词")

二、CANN图像内容审核

2.1 图像分类审核

使用CANN进行图像内容安全检测：

from transformers import AutoImageProcessor, AutoModelForImageClassification
from PIL import Image

class CANNImageModerator:
    """CANN图像内容审核器"""
    
    def __init__(self, model_name="google/vit-base-patch16-224"):
        self.device = "npu:0"
        print(f"正在加载图像审核模型到 {self.device}...")
        
        # 加载模型
        self.processor = AutoImageProcessor.from_pretrained(model_name)
        self.model = AutoModelForImageClassification.from_pretrained(
            model_name,
            torch_dtype=torch.float16
        ).to(self.device)
        
        self.model.eval()
        self._enable_cann_optimizations()
        
        print("图像审核模型加载完成")
    
    def _enable_cann_optimizations(self):
        """启用CANN优化"""
        for param in self.model.parameters():
            param.requires_grad = False
        
        torch_npu.npu.set_option({
            'ACL_OP_COMPILER_CACHE_MODE': '1'
        })
    
    def moderate_image(self, image, unsafe_categories=None):
        """审核单张图像"""
        if unsafe_categories is None:
            unsafe_categories = ['violence', 'nsfw', 'gore']
        
        # 处理图像
        inputs = self.processor(images=image, return_tensors="pt")
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1)
        
        # 获取top-5预测
        top5_probs, top5_indices = torch.topk(probs, 5)
        
        # 检查是否包含不安全类别
        is_safe = True
        detected_categories = []
        
        for prob, idx in zip(top5_probs[0], top5_indices[0]):
            label = self.model.config.id2label[idx.item()]
            
            if any(unsafe in label.lower() for unsafe in unsafe_categories):
                is_safe = False
                detected_categories.append({
                    'category': label,
                    'confidence': prob.item()
                })
        
        result = {
            'is_safe': is_safe,
            'detected_unsafe': detected_categories,
            'top_predictions': [
                {
                    'label': self.model.config.id2label[idx.item()],
                    'confidence': prob.item()
                }
                for prob, idx in zip(top5_probs[0], top5_indices[0])
            ]
        }
        
        return result
    
    def batch_moderate_images(self, images, batch_size=16):
        """批量审核图像"""
        results = []
        
        for i in range(0, len(images), batch_size):
            batch_images = images[i:i+batch_size]
            
            # 批量处理
            inputs = self.processor(images=batch_images, return_tensors="pt")
            inputs = {k: v.to(self.device) for k, v in inputs.items()}
            
            with torch.no_grad():
                outputs = self.model(**inputs)
                probs = torch.softmax(outputs.logits, dim=-1)
            
            # 处理每张图像的结果
            for prob_dist in probs:
                max_prob, pred_idx = torch.max(prob_dist, dim=0)
                label = self.model.config.id2label[pred_idx.item()]
                
                results.append({
                    'is_safe': 'unsafe' not in label.lower(),
                    'category': label,
                    'confidence': max_prob.item()
                })
        
        return results

# 使用示例
image_moderator = CANNImageModerator()

# 加载图像
image = Image.open("test_image.jpg")

# 审核图像
result = image_moderator.moderate_image(image)

print("图像审核结果:")
print(f"安全状态: {'✓ 安全' if result['is_safe'] else '✗ 不安全'}")

if not result['is_safe']:
    print("检测到的不安全内容:")
    for item in result['detected_unsafe']:
        print(f"  - {item['category']}: {item['confidence']:.2%}")

2.2 OCR文本提取与审核

从图像中提取文本并进行审核：

import easyocr

class CANNImageTextModerator:
    """CANN图像文本审核器"""
    
    def __init__(self):
        self.ocr_reader = easyocr.Reader(['ch_sim', 'en'])
        self.text_moderator = CANNTextModerator()
        print("图像文本审核器初始化完成")
    
    def extract_and_moderate(self, image):
        """提取图像中的文本并审核"""
        # OCR提取文本
        ocr_results = self.ocr_reader.readtext(image)
        
        # 合并所有文本
        extracted_texts = [text for (bbox, text, conf) in ocr_results]
        full_text = ' '.join(extracted_texts)
        
        if not full_text.strip():
            return {
                'has_text': False,
                'is_safe': True,
                'extracted_text': '',
                'moderation_result': None
            }
        
        # 审核提取的文本
        moderation_result = self.text_moderator.moderate_text(full_text)
        
        return {
            'has_text': True,
            'is_safe': moderation_result['is_safe'],
            'extracted_text': full_text,
            'moderation_result': moderation_result,
            'ocr_details': ocr_results
        }

# 使用示例
image_text_mod = CANNImageTextModerator()

# 审核包含文本的图像
image_with_text = Image.open("image_with_text.jpg")
result = image_text_mod.extract_and_moderate(image_with_text)

print("图像文本审核结果:")
print(f"包含文本: {result['has_text']}")
if result['has_text']:
    print(f"提取的文本: {result['extracted_text']}")
    print(f"安全状态: {'✓ 安全' if result['is_safe'] else '✗ 违规'}")

三、CANN多模态内容审核系统

3.1 综合审核Pipeline

构建完整的多模态内容审核流水线：

class CANNMultiModalModerator:
    """CANN多模态内容审核系统"""
    
    def __init__(self):
        self.text_moderator = CANNTextModerator()
        self.image_moderator = CANNImageModerator()
        self.image_text_moderator = CANNImageTextModerator()
        self.word_filter = CANNSensitiveWordFilter()
        
        print("多模态审核系统初始化完成")
    
    def moderate_content(self, content_type, content, **kwargs):
        """审核内容"""
        if content_type == 'text':
            return self._moderate_text_content(content)
        elif content_type == 'image':
            return self._moderate_image_content(content)
        elif content_type == 'mixed':
            return self._moderate_mixed_content(content['text'], content['image'])
        else:
            raise ValueError(f"不支持的内容类型: {content_type}")
    
    def _moderate_text_content(self, text):
        """审核纯文本内容"""
        # 敏感词检测
        has_sensitive, sensitive_words = self.word_filter.contains_sensitive_words(text)
        
        # AI模型审核
        ai_result = self.text_moderator.moderate_text(text)
        
        # 综合判断
        is_safe = ai_result['is_safe'] and not has_sensitive
        
        return {
            'content_type': 'text',
            'is_safe': is_safe,
            'ai_moderation': ai_result,
            'sensitive_words': sensitive_words if has_sensitive else [],
            'risk_level': self._calculate_risk_level(ai_result, has_sensitive)
        }
    
    def _moderate_image_content(self, image):
        """审核图像内容"""
        # 图像分类审核
        image_result = self.image_moderator.moderate_image(image)
        
        # 图像文本审核
        text_result = self.image_text_moderator.extract_and_moderate(image)
        
        # 综合判断
        is_safe = image_result['is_safe'] and text_result['is_safe']
        
        return {
            'content_type': 'image',
            'is_safe': is_safe,
            'image_moderation': image_result,
            'text_in_image': text_result,
            'risk_level': self._calculate_image_risk(image_result, text_result)
        }
    
    def _moderate_mixed_content(self, text, image):
        """审核图文混合内容"""
        text_result = self._moderate_text_content(text)
        image_result = self._moderate_image_content(image)
        
        is_safe = text_result['is_safe'] and image_result['is_safe']
        
        return {
            'content_type': 'mixed',
            'is_safe': is_safe,
            'text_moderation': text_result,
            'image_moderation': image_result,
            'overall_risk_level': max(
                text_result['risk_level'],
                image_result['risk_level']
            )
        }
    
    def _calculate_risk_level(self, ai_result, has_sensitive):
        """计算风险等级"""
        if has_sensitive:
            return 'high'
        
        if not ai_result['is_safe']:
            if ai_result['confidence'] > 0.9:
                return 'high'
            elif ai_result['confidence'] > 0.7:
                return 'medium'
            else:
                return 'low'
        
        return 'safe'
    
    def _calculate_image_risk(self, image_result, text_result):
        """计算图像风险等级"""
        if not image_result['is_safe'] or not text_result['is_safe']:
            return 'high'
        return 'safe'

# 使用示例
multi_moderator = CANNMultiModalModerator()

# 审核文本
text_result = multi_moderator.moderate_content('text', "This is a test message.")
print("文本审核:")
print(f"  安全: {text_result['is_safe']}")
print(f"  风险等级: {text_result['risk_level']}")

# 审核图像
image = Image.open("test.jpg")
image_result = multi_moderator.moderate_content('image', image)
print("\n图像审核:")
print(f"  安全: {image_result['is_safe']}")
print(f"  风险等级: {image_result['risk_level']}")

# 审核混合内容
mixed_result = multi_moderator.moderate_content(
    'mixed',
    {'text': "Check out this image", 'image': image}
)
print("\n混合内容审核:")
print(f"  安全: {mixed_result['is_safe']}")
print(f"  总体风险: {mixed_result['overall_risk_level']}")

3.2 实时审核服务

构建高性能的实时审核API服务：

from flask import Flask, request, jsonify
import base64
from io import BytesIO

app = Flask(__name__)
moderator = CANNMultiModalModerator()

@app.route('/moderate/text', methods=['POST'])
def moderate_text():
    """文本审核API"""
    data = request.json
    text = data.get('text', '')
    
    if not text:
        return jsonify({'error': 'No text provided'}), 400
    
    result = moderator.moderate_content('text', text)
    
    return jsonify(result)

@app.route('/moderate/image', methods=['POST'])
def moderate_image():
    """图像审核API"""
    if 'image' not in request.files:
        return jsonify({'error': 'No image provided'}), 400
    
    image_file = request.files['image']
    image = Image.open(image_file.stream)
    
    result = moderator.moderate_content('image', image)
    
    return jsonify(result)

@app.route('/moderate/batch', methods=['POST'])
def moderate_batch():
    """批量审核API"""
    data = request.json
    items = data.get('items', [])
    
    results = []
    for item in items:
        content_type = item.get('type')
        content = item.get('content')
        
        try:
            result = moderator.moderate_content(content_type, content)
            results.append(result)
        except Exception as e:
            results.append({
                'error': str(e),
                'is_safe': False
            })
    
    return jsonify({
        'total': len(items),
        'results': results
    })

@app.route('/health', methods=['GET'])
def health_check():
    """健康检查"""
    return jsonify({'status': 'healthy'})

if __name__ == '__main__':
    print("CANN内容审核服务启动...")
    app.run(host='0.0.0.0', port=8080)

四、性能优化与监控

4.1 审核性能优化

优化大规模内容审核的性能：

import time
from concurrent.futures import ThreadPoolExecutor

class CANNOptimizedModerator:
    """优化的内容审核器"""
    
    def __init__(self, num_workers=4):
        self.moderator = CANNMultiModalModerator()
        self.num_workers = num_workers
        self.executor = ThreadPoolExecutor(max_workers=num_workers)
    
    def parallel_moderate(self, items):
        """并行审核多个内容"""
        futures = []
        
        for item in items:
            future = self.executor.submit(
                self.moderator.moderate_content,
                item['type'],
                item['content']
            )
            futures.append(future)
        
        # 收集结果
        results = [future.result() for future in futures]
        
        return results
    
    def benchmark_performance(self, num_items=100):
        """性能基准测试"""
        # 生成测试数据
        test_items = [
            {'type': 'text', 'content': f'Test message {i}'}
            for i in range(num_items)
        ]
        
        # 串行处理
        start = time.time()
        for item in test_items:
            _ = self.moderator.moderate_content(item['type'], item['content'])
        serial_time = time.time() - start
        
        # 并行处理
        start = time.time()
        _ = self.parallel_moderate(test_items)
        parallel_time = time.time() - start
        
        print(f"性能测试结果 ({num_items} 条内容):")
        print(f"  串行处理: {serial_time:.2f}秒")
        print(f"  并行处理: {parallel_time:.2f}秒")
        print(f"  加速比: {serial_time/parallel_time:.2f}x")
        print(f"  吞吐量: {num_items/parallel_time:.2f} 条/秒")

# 性能测试
opt_moderator = CANNOptimizedModerator(num_workers=4)
opt_moderator.benchmark_performance(num_items=100)

4.2 审核统计与监控

实现审核结果的统计和监控：

from collections import defaultdict
import json

class CANNModerationMonitor:
    """内容审核监控系统"""
    
    def __init__(self):
        self.stats = defaultdict(int)
        self.violation_log = []
    
    def log_moderation(self, result):
        """记录审核结果"""
        # 更新统计
        self.stats['total'] += 1
        
        if result['is_safe']:
            self.stats['safe'] += 1
        else:
            self.stats['unsafe'] += 1
            self.violation_log.append(result)
        
        # 按类型统计
        content_type = result.get('content_type', 'unknown')
        self.stats[f'{content_type}_total'] += 1
        
        if not result['is_safe']:
            self.stats[f'{content_type}_unsafe'] += 1
    
    def get_statistics(self):
        """获取统计信息"""
        total = self.stats['total']
        if total == 0:
            return {}
        
        return {
            'total_processed': total,
            'safe_count': self.stats['safe'],
            'unsafe_count': self.stats['unsafe'],
            'safety_rate': self.stats['safe'] / total * 100,
            'violation_rate': self.stats['unsafe'] / total * 100,
            'by_type': {
                'text': {
                    'total': self.stats['text_total'],
                    'unsafe': self.stats['text_unsafe']
                },
                'image': {
                    'total': self.stats['image_total'],
                    'unsafe': self.stats['image_unsafe']
                }
            }
        }
    
    def export_report(self, filename='moderation_report.json'):
        """导出审核报告"""
        report = {
            'statistics': self.get_statistics(),
            'recent_violations': self.violation_log[-100:]  # 最近100条违规
        }
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        
        print(f"审核报告已导出: {filename}")

# 使用示例
monitor = CANNModerationMonitor()

# 模拟审核并记录
moderator = CANNMultiModalModerator()

test_texts = ["Normal text", "Another message", "Test content"]
for text in test_texts:
    result = moderator.moderate_content('text', text)
    monitor.log_moderation(result)

# 查看统计
stats = monitor.get_statistics()
print("\n审核统计:")
print(f"  总处理量: {stats['total_processed']}")
print(f"  安全率: {stats['safety_rate']:.2f}%")
print(f"  违规率: {stats['violation_rate']:.2f}%")

# 导出报告
monitor.export_report()

总结

本文介绍了如何使用CANN平台构建多模态内容审核系统。从文本审核、图像审核到综合的多模态审核，CANN提供了高效、准确的内容安全检测解决方案。

核心内容：

• CANN环境下文本和图像审核模型的部署
• 敏感词过滤、OCR文本提取等辅助功能
• 完整的多模态审核流水线和实时服务
• 并行处理优化和审核监控系统

通过CANN的加速能力，开发者可以构建高性能的内容审核系统，为AIGC应用的安全合规提供技术保障。