ميزانيات التأخير ومقايضات الأداء

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

واقع التأخير

تفصيل تأخير حواجز الإنتاج النموذجي:

المكون	نطاق التأخير	ملاحظات
Regex/قائمة الحظر	< 1 مللي ثانية	مهمل
تشابه التضمين	5-15 مللي ثانية	يعتمد على قاعدة بيانات المتجهات
مصنف صغير (DistilBERT)	10-30 مللي ثانية	استدلال CPU
LlamaGuard 3 1B	50-100 مللي ثانية	يتطلب GPU
LlamaGuard 3 8B	100-300 مللي ثانية	أكثر دقة
NeMo Guardrails (مع LLM)	200-500 مللي ثانية	يتضمن الاستدلال
ShieldGemma 27B الكامل	300-800 مللي ثانية	الأكثر شمولاً

قاعدة الميزانية: معظم واجهات API الإنتاجية تستهدف 2-3 ثانية وقت استجابة إجمالي. الحواجز يجب أن تستهلك < 20% من تلك الميزانية (~400-600 مللي ثانية إجمالي).

استراتيجية توزيع ميزانية التأخير

from dataclasses import dataclass
from typing import Callable, Awaitable
import asyncio
import time

@dataclass
class LatencyBudget:
    """إدارة الحواجز ضمن قيود التأخير."""
    total_budget_ms: int = 500
    edge_budget_ms: int = 10
    classification_budget_ms: int = 150
    output_budget_ms: int = 100

    async def run_with_budget(
        self,
        func: Callable[[], Awaitable],
        budget_ms: int,
        fallback: Callable[[], Awaitable] = None
    ):
        """تشغيل الدالة مع مهلة، استخدام الاحتياطي إذا تم التجاوز."""
        try:
            return await asyncio.wait_for(
                func(),
                timeout=budget_ms / 1000
            )
        except asyncio.TimeoutError:
            if fallback:
                return await fallback()
            raise

# مثال: التصنيف مع الاحتياطي
async def run_classification(user_input: str, budget: LatencyBudget):
    async def primary_check():
        # LlamaGuard 8B - دقيق لكن بطيء
        return await llamaguard_8b.classify(user_input)

    async def fallback_check():
        # DistilBERT - سريع لكن أقل دقة
        return await distilbert_toxic.classify(user_input)

    return await budget.run_with_budget(
        primary_check,
        budget.classification_budget_ms,
        fallback=fallback_check
    )

المعالجة المتوازية مقابل التسلسلية

المفتاح للحواجز السريعة: تشغيل الفحوصات المستقلة بالتوازي.

import asyncio
from typing import List, Dict, Any

async def parallel_input_checks(user_input: str) -> Dict[str, Any]:
    """تشغيل فحوصات مدخلات متعددة بشكل متزامن."""
    # كل هذه الفحوصات مستقلة
    results = await asyncio.gather(
        toxicity_classifier.check(user_input),
        pii_detector.scan(user_input),
        injection_classifier.check(user_input),
        blocklist_matcher.check(user_input),
        return_exceptions=True
    )

    return {
        "toxicity": results[0] if not isinstance(results[0], Exception) else None,
        "pii": results[1] if not isinstance(results[1], Exception) else None,
        "injection": results[2] if not isinstance(results[2], Exception) else None,
        "blocklist": results[3] if not isinstance(results[3], Exception) else None,
    }

# مثال مقارنة التوقيت
async def demo_parallel_speedup():
    # تسلسلي: 50 + 30 + 40 + 5 = 125 مللي ثانية
    # متوازي: max(50, 30, 40, 5) = 50 مللي ثانية
    # تسريع: 2.5x
    pass

استراتيجية التصنيف المتدرج

استخدم النماذج السريعة أولاً، والنماذج المكلفة فقط عند الحاجة:

from enum import Enum

class SafetyDecision(Enum):
    SAFE = "safe"
    UNSAFE = "unsafe"
    UNCERTAIN = "uncertain"

async def tiered_safety_check(user_input: str) -> tuple[SafetyDecision, float]:
    """تصنيف ثلاثي المستويات: سريع → متوسط → شامل."""

    # المستوى 1: فحص regex فائق السرعة (< 1 مللي ثانية)
    if contains_obvious_attack_pattern(user_input):
        return SafetyDecision.UNSAFE, 0.0

    # المستوى 2: مصنف سريع (10-30 مللي ثانية)
    fast_score = await distilbert_toxic.score(user_input)
    if fast_score > 0.95:  # ثقة عالية غير آمن
        return SafetyDecision.UNSAFE, fast_score
    if fast_score < 0.1:   # ثقة عالية آمن
        return SafetyDecision.SAFE, fast_score

    # المستوى 3: تصعيد الحالات غير المؤكدة فقط (100-300 مللي ثانية)
    # هذا يُستدعى لـ ~10-20% من الطلبات
    detailed_result = await llamaguard_8b.classify(user_input)
    return (
        SafetyDecision.UNSAFE if detailed_result.unsafe else SafetyDecision.SAFE,
        detailed_result.confidence
    )

def contains_obvious_attack_pattern(text: str) -> bool:
    """مطابقة أنماط بمستوى الميكروثانية."""
    patterns = [
        "ignore all previous",
        "disregard your instructions",
        "you are now in developer mode",
    ]
    text_lower = text.lower()
    return any(p in text_lower for p in patterns)

تقنيات تحسين الأداء

التقنية	تقليل التأخير	التنفيذ
التجميع	50-70%	تجميع طلبات متعددة لاستدلال GPU
التخزين المؤقت	90%+	تخزين النتائج للمدخلات المتكررة/المشابهة
التكميم	30-50%	استخدام نماذج INT8/FP16
تقطير النموذج	60-80%	نماذج أصغر مدربة على نماذج أكبر
الإنهاء المبكر	متغير	التوقف عند أول نتيجة حاسمة

from functools import lru_cache
import hashlib

# تخزين مؤقت بسيط للمدخلات المتكررة
@lru_cache(maxsize=10000)
def cached_blocklist_check(input_hash: str) -> bool:
    # الفحص الفعلي يتم عند عدم وجود تخزين مؤقت
    pass

def check_with_cache(user_input: str) -> bool:
    input_hash = hashlib.md5(user_input.encode()).hexdigest()
    return cached_blocklist_check(input_hash)

نقطة رئيسية: مع التصنيف المتدرج، 80% من الطلبات تكتمل في < 50 مللي ثانية، بينما المدخلات المشبوهة فقط تُطلق النماذج المكلفة.

التالي: تصميم خطوط أنابيب تصفية متعددة الطبقات للحماية الشاملة. :::