المراقبة والمقاييس

تتطلب حواجز الإنتاج مراقبة شاملة لتتبع السلامة والأداء والتأثير التجاري. يغطي هذا الدرس المقاييس الأساسية واستراتيجيات المراقبة.

مقاييس الحواجز الرئيسية

مقاييس السلامة

المقياس	الوصف	الهدف
معدل الحظر	% من الطلبات المحظورة	< 5% للإنتاج
معدل الإيجابيات الخاطئة	محتوى آمن محظور بشكل خاطئ	< 1%
معدل السلبيات الخاطئة	محتوى غير آمن تم تمريره	< 0.1%
توزيع الفئات	تفصيل الانتهاكات حسب النوع	يختلف
معدل التصعيد	% يتطلب مراجعة بشرية	< 2%

مقاييس الأداء

المقياس	الوصف	الهدف
تأخير P50	وقت الحاجز المتوسط	< 50 مللي ثانية
تأخير P99	المئين 99	< 200 مللي ثانية
الإنتاجية	الطلبات في الثانية	يختلف
معدل الخطأ	فشل الحواجز	< 0.1%
معدل المهلة	مهلات الحواجز	< 0.01%

تنفيذ مقاييس Prometheus

from prometheus_client import Counter, Histogram, Gauge
import time
from functools import wraps

# تعريف المقاييس
GUARDRAIL_REQUESTS = Counter(
    'guardrail_requests_total',
    'إجمالي طلبات الحواجز',
    ['rail_type', 'result']
)

GUARDRAIL_LATENCY = Histogram(
    'guardrail_latency_seconds',
    'تأخير معالجة الحواجز',
    ['rail_type'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)

GUARDRAIL_BLOCKS = Counter(
    'guardrail_blocks_total',
    'إجمالي الطلبات المحظورة',
    ['rail_type', 'category']
)

ACTIVE_GUARDRAIL_CHECKS = Gauge(
    'guardrail_active_checks',
    'فحوصات الحواجز قيد التشغيل حالياً',
    ['rail_type']
)

class MetricsMiddleware:
    """برمجية وسيطة لتتبع مقاييس الحواجز."""

    def __init__(self, rail_type: str):
        self.rail_type = rail_type

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            ACTIVE_GUARDRAIL_CHECKS.labels(rail_type=self.rail_type).inc()
            start_time = time.time()

            try:
                result = await func(*args, **kwargs)

                # تتبع النتيجة
                outcome = "blocked" if result.blocked else "passed"
                GUARDRAIL_REQUESTS.labels(
                    rail_type=self.rail_type,
                    result=outcome
                ).inc()

                if result.blocked:
                    for category in result.categories:
                        GUARDRAIL_BLOCKS.labels(
                            rail_type=self.rail_type,
                            category=category
                        ).inc()

                return result

            except Exception as e:
                GUARDRAIL_REQUESTS.labels(
                    rail_type=self.rail_type,
                    result="error"
                ).inc()
                raise

            finally:
                duration = time.time() - start_time
                GUARDRAIL_LATENCY.labels(
                    rail_type=self.rail_type
                ).observe(duration)
                ACTIVE_GUARDRAIL_CHECKS.labels(rail_type=self.rail_type).dec()

        return wrapper

# الاستخدام
@MetricsMiddleware(rail_type="toxicity")
async def check_toxicity(text: str):
    # منطق فحص السمية
    pass

التسجيل المنظم

import structlog
from datetime import datetime
from typing import Dict, Any
import json

# تكوين التسجيل المنظم
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
)

logger = structlog.get_logger()

class GuardrailLogger:
    """التسجيل المنظم لأحداث الحواجز."""

    def log_check(
        self,
        request_id: str,
        rail_type: str,
        input_text: str,
        result: Dict[str, Any],
        latency_ms: float
    ):
        """تسجيل نتيجة فحص الحاجز."""
        logger.info(
            "guardrail_check",
            request_id=request_id,
            rail_type=rail_type,
            input_length=len(input_text),
            result=result.get("decision"),
            categories=result.get("categories", []),
            confidence=result.get("confidence"),
            latency_ms=latency_ms,
            timestamp=datetime.utcnow().isoformat()
        )

    def log_block(
        self,
        request_id: str,
        rail_type: str,
        reason: str,
        categories: list,
        input_preview: str = None
    ):
        """تسجيل طلب محظور."""
        logger.warning(
            "guardrail_block",
            request_id=request_id,
            rail_type=rail_type,
            reason=reason,
            categories=categories,
            input_preview=input_preview[:100] if input_preview else None
        )

    def log_error(
        self,
        request_id: str,
        rail_type: str,
        error: Exception,
        fallback_action: str
    ):
        """تسجيل خطأ الحاجز."""
        logger.error(
            "guardrail_error",
            request_id=request_id,
            rail_type=rail_type,
            error_type=type(error).__name__,
            error_message=str(error),
            fallback_action=fallback_action
        )

لوحة القيادة الفورية

from dataclasses import dataclass
from datetime import datetime, timedelta
from collections import defaultdict
import asyncio

@dataclass
class MetricWindow:
    """نافذة متحركة لتجميع المقاييس."""
    window_seconds: int = 60
    buckets: dict = None

    def __post_init__(self):
        self.buckets = defaultdict(list)

    def record(self, key: str, value: float):
        now = datetime.now()
        self.buckets[key].append((now, value))
        self._cleanup(key)

    def _cleanup(self, key: str):
        cutoff = datetime.now() - timedelta(seconds=self.window_seconds)
        self.buckets[key] = [
            (ts, v) for ts, v in self.buckets[key]
            if ts > cutoff
        ]

    def get_stats(self, key: str) -> dict:
        values = [v for _, v in self.buckets[key]]
        if not values:
            return {"count": 0, "avg": 0, "p99": 0}

        values.sort()
        return {
            "count": len(values),
            "avg": sum(values) / len(values),
            "p50": values[len(values) // 2],
            "p99": values[int(len(values) * 0.99)]
        }

class RealTimeDashboard:
    """لوحة قيادة المقاييس الفورية."""

    def __init__(self):
        self.latency = MetricWindow(window_seconds=60)
        self.block_rates = MetricWindow(window_seconds=300)
        self.error_counts = defaultdict(int)

    def record_check(
        self,
        rail_type: str,
        latency_ms: float,
        blocked: bool
    ):
        self.latency.record(rail_type, latency_ms)
        self.block_rates.record(rail_type, 1 if blocked else 0)

    def record_error(self, rail_type: str):
        self.error_counts[rail_type] += 1

    def get_dashboard_data(self) -> dict:
        """الحصول على مقاييس لوحة القيادة الحالية."""
        data = {}
        for rail_type in self.latency.buckets:
            latency_stats = self.latency.get_stats(rail_type)
            block_stats = self.block_rates.get_stats(rail_type)

            data[rail_type] = {
                "latency_ms": {
                    "avg": latency_stats["avg"],
                    "p50": latency_stats["p50"],
                    "p99": latency_stats["p99"]
                },
                "requests_per_minute": latency_stats["count"],
                "block_rate": block_stats["avg"] * 100,
                "errors": self.error_counts[rail_type]
            }
        return data

قواعد التنبيه

# prometheus_alerts.yml
groups:
  - name: guardrails
    rules:
      - alert: HighGuardrailLatency
        expr: histogram_quantile(0.99, rate(guardrail_latency_seconds_bucket[5m])) > 0.2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "تأخير P99 للحاجز أعلى من 200 مللي ثانية"

      - alert: HighBlockRate
        expr: rate(guardrail_blocks_total[5m]) / rate(guardrail_requests_total[5m]) > 0.1
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "معدل الحظر أعلى من 10%"

      - alert: GuardrailErrors
        expr: rate(guardrail_requests_total{result="error"}[5m]) > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "معدل خطأ الحاجز أعلى من 1%"

      - alert: UnusualViolationSpike
        expr: rate(guardrail_blocks_total{category="hate_speech"}[5m]) > 2 * avg_over_time(rate(guardrail_blocks_total{category="hate_speech"}[5m])[1h:5m])
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "ارتفاع غير عادي في انتهاكات خطاب الكراهية"

نصيحة المراقبة: تتبع كل من مقاييس السلامة والأداء. الحاجز الذي يكون دقيقاً 100% لكنه يضيف 5 ثوان من التأخير غير قابل للاستخدام. وازن بين الشمولية وتجربة المستخدم.

التالي: اختبار A/B لتكوينات الحواجز. :::