مصنفات السمية السريعة

اكتشاف السمية السريع ضروري للطبقة الأولى من تصفية المدخلات. يغطي هذا الدرس تنفيذ مصنفات خفيفة الوزن يمكنها معالجة المدخلات في أقل من 30 مللي ثانية مع الحفاظ على دقة مقبولة.

نماذج تصنيف السمية

النموذج	الحجم	التأخير (CPU)	الدقة	حالة الاستخدام
toxic-bert (مقطر)	66MB	15-30 مللي ثانية	~85%	مرور أول سريع
unitary/toxic-bert	420MB	50-100 مللي ثانية	~90%	متوازن
HateBERT	420MB	50-100 مللي ثانية	~88%	تركيز خطاب الكراهية
detoxify	1.3GB	100-200 مللي ثانية	~93%	دقة عالية

تنفيذ DistilBERT للسمية

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
from typing import Dict, List
import torch

class FastToxicityClassifier:
    """مصنف سمية خفيف الوزن لتصفية مدخلات الإنتاج."""

    def __init__(self, model_name: str = "martin-ha/toxic-comment-model"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.to(self.device)
        self.model.eval()

    @torch.no_grad()
    def classify(self, text: str) -> Dict[str, float]:
        """تصنيف نص واحد للسمية."""
        inputs = self.tokenizer(
            text,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        ).to(self.device)

        outputs = self.model(**inputs)
        probs = torch.softmax(outputs.logits, dim=-1)

        return {
            "toxic": float(probs[0][1]),
            "non_toxic": float(probs[0][0])
        }

    @torch.no_grad()
    def batch_classify(self, texts: List[str], batch_size: int = 32) -> List[Dict]:
        """تصنيف دفعي للإنتاجية."""
        results = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            inputs = self.tokenizer(
                batch,
                return_tensors="pt",
                truncation=True,
                max_length=512,
                padding=True
            ).to(self.device)

            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1)

            for j in range(len(batch)):
                results.append({
                    "toxic": float(probs[j][1]),
                    "non_toxic": float(probs[j][0])
                })

        return results

# الاستخدام
classifier = FastToxicityClassifier()
result = classifier.classify("هذه رسالة اختبار")
print(f"درجة السمية: {result['toxic']:.2%}")

السمية متعددة الفئات مع Detoxify

from detoxify import Detoxify

class DetailedToxicityClassifier:
    """اكتشاف سمية متعدد الفئات."""

    def __init__(self, model_type: str = "original"):
        # model_type: "original", "unbiased", "multilingual"
        self.model = Detoxify(model_type)
        self.categories = [
            "toxicity", "severe_toxicity", "obscene",
            "threat", "insult", "identity_attack"
        ]

    def classify(self, text: str) -> Dict[str, float]:
        """الحصول على درجات لجميع فئات السمية."""
        return self.model.predict(text)

    def check(self, text: str, thresholds: Dict[str, float] = None) -> bool:
        """التحقق إذا كان النص يتجاوز أي عتبة."""
        default_thresholds = {
            "toxicity": 0.8,
            "severe_toxicity": 0.5,
            "threat": 0.5,
            "identity_attack": 0.5,
        }
        thresholds = thresholds or default_thresholds

        scores = self.classify(text)
        for category, threshold in thresholds.items():
            if scores.get(category, 0) > threshold:
                return True  # سام
        return False  # آمن

# الاستخدام
classifier = DetailedToxicityClassifier()
scores = classifier.classify("نص عينة للتحليل")
# يُرجع: {'toxicity': 0.02, 'severe_toxicity': 0.001, ...}

تحسين ONNX للإنتاج

تحويل النماذج إلى ONNX لتسريع 2-3x:

from transformers import AutoTokenizer, AutoModelForSequenceClassification
from optimum.onnxruntime import ORTModelForSequenceClassification
import numpy as np

class ONNXToxicityClassifier:
    """مصنف سمية محسن ONNX للإنتاج."""

    def __init__(self, model_name: str = "martin-ha/toxic-comment-model"):
        # تصدير وتحميل كـ ONNX
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = ORTModelForSequenceClassification.from_pretrained(
            model_name,
            export=True  # تصدير تلقائي لـ ONNX
        )

    def classify(self, text: str) -> float:
        """تصنيف فائق السرعة مع ONNX."""
        inputs = self.tokenizer(
            text,
            return_tensors="np",
            truncation=True,
            max_length=512
        )

        outputs = self.model(**inputs)
        probs = self._softmax(outputs.logits[0])
        return float(probs[1])  # احتمالية السمية

    def _softmax(self, x):
        e_x = np.exp(x - np.max(x))
        return e_x / e_x.sum()

# مقارنة الأداء
import time

def benchmark_classifiers():
    text = "هذه رسالة اختبار لتصنيف السمية"

    # PyTorch القياسي
    torch_classifier = FastToxicityClassifier()
    start = time.time()
    for _ in range(100):
        torch_classifier.classify(text)
    torch_time = (time.time() - start) / 100 * 1000

    # ONNX
    onnx_classifier = ONNXToxicityClassifier()
    start = time.time()
    for _ in range(100):
        onnx_classifier.classify(text)
    onnx_time = (time.time() - start) / 100 * 1000

    print(f"PyTorch: {torch_time:.2f} مللي ثانية لكل استدلال")
    print(f"ONNX: {onnx_time:.2f} مللي ثانية لكل استدلال")
    print(f"التسريع: {torch_time/onnx_time:.2f}x")

التكامل مع خط أنابيب الحواجز

from dataclasses import dataclass
from enum import Enum

class ToxicityDecision(Enum):
    SAFE = "safe"
    FLAGGED = "flagged"
    BLOCKED = "blocked"

@dataclass
class ToxicityResult:
    decision: ToxicityDecision
    score: float
    categories: Dict[str, float]

async def toxicity_filter_layer(
    user_input: str,
    block_threshold: float = 0.9,
    flag_threshold: float = 0.5
) -> ToxicityResult:
    """طبقة مرشح السمية الإنتاجية."""
    classifier = FastToxicityClassifier()

    # تصنيف سريع
    result = classifier.classify(user_input)
    toxic_score = result["toxic"]

    # منطق القرار
    if toxic_score > block_threshold:
        return ToxicityResult(
            decision=ToxicityDecision.BLOCKED,
            score=toxic_score,
            categories=result
        )
    elif toxic_score > flag_threshold:
        return ToxicityResult(
            decision=ToxicityDecision.FLAGGED,
            score=toxic_score,
            categories=result
        )

    return ToxicityResult(
        decision=ToxicityDecision.SAFE,
        score=toxic_score,
        categories=result
    )

نصيحة إنتاجية: استخدم النماذج المحسنة ONNX لنشر CPU. لـ GPU، اجمع طلبات متعددة معاً لإنتاجية أعلى. استهدف تأخير < 30 مللي ثانية لفحص السمية الأول.

التالي: بناء مدققات مدخلات مخصصة للمتطلبات الخاصة بالمجال. :::