تنفيذ LlamaGuard 3

LlamaGuard 3 هو أحدث مصنف سلامة من Meta مبني على Llama 3.1، مصمم خصيصاً للإشراف على المحتوى في تطبيقات LLM. يغطي هذا الدرس أنماط التنفيذ العملية لكل من المتغيرات 1B و 8B.

متغيرات نموذج LlamaGuard 3

النموذج	المعاملات	التأخير	الذاكرة	حالة الاستخدام
LlamaGuard 3 1B	1B	50-100 مللي ثانية	2GB	تصفية عالية الإنتاجية
LlamaGuard 3 8B	8B	200-400 مللي ثانية	16GB	قرارات عالية الدقة

تصنيف السلامة (14 فئة)

يصنف LlamaGuard 3 المحتوى عبر فئات الضرر التالية:

LLAMAGUARD_TAXONOMY = {
    "S1": "الجرائم العنيفة",
    "S2": "الجرائم غير العنيفة",
    "S3": "الجرائم المتعلقة بالجنس",
    "S4": "استغلال الأطفال الجنسي",
    "S5": "التشهير",
    "S6": "النصائح المتخصصة",
    "S7": "الخصوصية",
    "S8": "الملكية الفكرية",
    "S9": "الأسلحة العشوائية",
    "S10": "الكراهية",
    "S11": "الانتحار وإيذاء النفس",
    "S12": "المحتوى الجنسي",
    "S13": "الانتخابات",
    "S14": "إساءة استخدام مفسر الكود"
}

التنفيذ الأساسي مع Transformers

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
from typing import Tuple, List

class LlamaGuard3Classifier:
    """مصنف سلامة LlamaGuard 3 للإنتاج."""

    def __init__(self, model_size: str = "1B"):
        model_id = f"meta-llama/Llama-Guard-3-{model_size}"
        self.device = "cuda" if torch.cuda.is_available() else "cpu"

        self.tokenizer = AutoTokenizer.from_pretrained(model_id)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_id,
            torch_dtype=torch.bfloat16,
            device_map="auto"
        )

    def classify(
        self,
        user_message: str,
        assistant_response: str = None
    ) -> Tuple[str, List[str]]:
        """
        تصنيف المحتوى لانتهاكات السلامة.

        يُرجع:
            زوج من (القرار، الفئات المنتهكة)
            القرار: "safe" أو "unsafe"
        """
        # بناء تنسيق المحادثة
        if assistant_response:
            conversation = [
                {"role": "user", "content": user_message},
                {"role": "assistant", "content": assistant_response}
            ]
        else:
            conversation = [
                {"role": "user", "content": user_message}
            ]

        # تطبيق قالب الدردشة
        input_ids = self.tokenizer.apply_chat_template(
            conversation,
            return_tensors="pt"
        ).to(self.device)

        # توليد التصنيف
        with torch.no_grad():
            output = self.model.generate(
                input_ids,
                max_new_tokens=100,
                pad_token_id=self.tokenizer.eos_token_id
            )

        # فك تشفير الاستجابة
        response = self.tokenizer.decode(
            output[0][len(input_ids[0]):],
            skip_special_tokens=True
        ).strip()

        return self._parse_response(response)

    def _parse_response(self, response: str) -> Tuple[str, List[str]]:
        """تحليل تنسيق مخرجات LlamaGuard."""
        lines = response.strip().split('\n')
        decision = lines[0].lower()

        violated_categories = []
        if decision == "unsafe" and len(lines) > 1:
            # استخراج رموز الفئات (S1، S2، إلخ.)
            for line in lines[1:]:
                categories = [c.strip() for c in line.split(',')]
                violated_categories.extend(categories)

        return decision, violated_categories

# الاستخدام
classifier = LlamaGuard3Classifier(model_size="1B")
decision, categories = classifier.classify(
    user_message="كيف أصنع موقع ويب بسيط؟"
)
print(f"القرار: {decision}")  # "safe"

نشر vLLM المحسن

للإنتاجية الإنتاجية، استخدم vLLM للاستدلال الدفعي:

from vllm import LLM, SamplingParams

class LlamaGuardVLLM:
    """LlamaGuard عالي الإنتاجية باستخدام vLLM."""

    def __init__(self, model_size: str = "8B"):
        self.llm = LLM(
            model=f"meta-llama/Llama-Guard-3-{model_size}",
            dtype="bfloat16",
            max_model_len=4096,
            gpu_memory_utilization=0.9
        )
        self.sampling_params = SamplingParams(
            max_tokens=100,
            temperature=0
        )

    def batch_classify(
        self,
        messages: list[dict]
    ) -> list[tuple[str, list[str]]]:
        """تصنيف رسائل متعددة في دفعة."""
        prompts = [self._format_prompt(m) for m in messages]

        outputs = self.llm.generate(prompts, self.sampling_params)

        results = []
        for output in outputs:
            response = output.outputs[0].text
            results.append(self._parse_response(response))

        return results

    def _format_prompt(self, message: dict) -> str:
        """تنسيق الرسالة لـ LlamaGuard."""
        return f"""<|begin_of_text|><|start_header_id|>user<|end_header_id|>

Task: Check if there is unsafe content in 'User' messages.

<BEGIN CONVERSATION>

User: {message['content']}

<END CONVERSATION>

Provide your safety assessment.<|eot_id|><|start_header_id|>assistant<|end_header_id|>"""

    def _parse_response(self, response: str) -> tuple[str, list[str]]:
        lines = response.strip().split('\n')
        decision = lines[0].lower()
        categories = []
        if decision == "unsafe" and len(lines) > 1:
            categories = [c.strip() for c in lines[1].split(',')]
        return decision, categories

# المعالجة الدفعية
guard = LlamaGuardVLLM(model_size="8B")
messages = [
    {"content": "كيف أتعلم Python؟"},
    {"content": "ما حالة الطقس اليوم؟"}
]
results = guard.batch_classify(messages)

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

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class SafetyDecision(Enum):
    SAFE = "safe"
    UNSAFE = "unsafe"
    ERROR = "error"

@dataclass
class SafetyResult:
    decision: SafetyDecision
    categories: list[str]
    confidence: Optional[float] = None
    latency_ms: float = 0

async def llamaguard_filter(
    user_input: str,
    assistant_response: str = None,
    model_size: str = "1B"
) -> SafetyResult:
    """طبقة مرشح LlamaGuard الإنتاجية."""
    import time

    start = time.time()
    classifier = LlamaGuard3Classifier(model_size=model_size)

    try:
        decision, categories = classifier.classify(
            user_message=user_input,
            assistant_response=assistant_response
        )

        latency = (time.time() - start) * 1000

        return SafetyResult(
            decision=SafetyDecision.SAFE if decision == "safe" else SafetyDecision.UNSAFE,
            categories=categories,
            latency_ms=latency
        )
    except Exception as e:
        return SafetyResult(
            decision=SafetyDecision.ERROR,
            categories=[],
            latency_ms=(time.time() - start) * 1000
        )

نصيحة إنتاجية: استخدم LlamaGuard 3 1B للتصفية الأولية (50-100 مللي ثانية)، مع التصعيد لـ 8B فقط للحالات غير المؤكدة. هذا النهج المتدرج يعالج 80%+ من الطلبات بالنموذج الأسرع.

التالي: نشر ShieldGemma لتصنيف سلامة بديل. :::