اكتشاف وإخفاء PII باستخدام Presidio

Microsoft Presidio هو إطار عمل مفتوح المصدر لاكتشاف وإخفاء هوية البيانات الحساسة في النص. يغطي هذا الدرس دمج Presidio في خط أنابيب حواجز LLM لحماية PII الإنتاجية.

لماذا Presidio لتطبيقات LLM

تواجه تطبيقات LLM خطرين حرجين لـ PII:

1. تسرب المدخلات: المستخدمون يشاركون عن طريق الخطأ بيانات حساسة لا يجب معالجتها
2. تسرب المخرجات: قد تهلوس LLMs أو تسرب بيانات تدريب تحتوي PII

Presidio يعالج كليهما مع اكتشاف كيانات سريع وقابل للتخصيص.

التثبيت والإعداد

# تثبيت مكونات Presidio
# pip install presidio-analyzer presidio-anonymizer

from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

# تهيئة المحركات
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

# اختبار الاكتشاف الأساسي
text = "Contact John Smith at john.smith@company.com or 555-123-4567"
results = analyzer.analyze(text=text, language="en")

for result in results:
    print(f"{result.entity_type}: {text[result.start:result.end]} (score: {result.score:.2f})")
# المخرجات:
# PERSON: John Smith (score: 0.85)
# EMAIL_ADDRESS: john.smith@company.com (score: 1.00)
# PHONE_NUMBER: 555-123-4567 (score: 0.75)

أنواع الكيانات لتطبيقات LLM

Presidio يدعم العديد من أنواع الكيانات. اضبط حسب متطلباتك:

# الكيانات الشائعة للصناعات المختلفة
HEALTHCARE_ENTITIES = [
    "PERSON", "DATE_TIME", "PHONE_NUMBER", "EMAIL_ADDRESS",
    "MEDICAL_LICENSE", "US_SSN", "US_DRIVER_LICENSE"
]

FINANCE_ENTITIES = [
    "PERSON", "CREDIT_CARD", "US_BANK_NUMBER", "IBAN_CODE",
    "US_SSN", "EMAIL_ADDRESS", "PHONE_NUMBER"
]

GENERAL_ENTITIES = [
    "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
    "CREDIT_CARD", "IP_ADDRESS", "LOCATION"
]

# التحليل مع كيانات محددة
results = analyzer.analyze(
    text=user_input,
    entities=GENERAL_ENTITIES,
    language="en"
)

استراتيجيات إخفاء الهوية الإنتاجية

حالات الاستخدام المختلفة تتطلب أساليب إخفاء هوية مختلفة:

from presidio_anonymizer.entities import OperatorConfig

class PIIAnonymizer:
    """إخفاء هوية PII إنتاجي مع استراتيجيات متعددة."""

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def mask(self, text: str) -> str:
        """استبدال PII بقيم مخفية."""
        results = self.analyzer.analyze(text=text, language="en")

        return self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "DEFAULT": OperatorConfig("replace", {"new_value": "[محذوف]"}),
                "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[بريد]"}),
                "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[هاتف]"}),
                "CREDIT_CARD": OperatorConfig("replace", {"new_value": "[بطاقة]"}),
            }
        ).text

    def hash_pii(self, text: str) -> str:
        """استبدال PII بقيم مجزأة (قابلة للعكس بالمفتاح)."""
        results = self.analyzer.analyze(text=text, language="en")

        return self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "DEFAULT": OperatorConfig("hash", {"hash_type": "sha256"})
            }
        ).text

    def encrypt_pii(self, text: str, key: str) -> str:
        """تشفير PII (قابل للعكس بالكامل)."""
        results = self.analyzer.analyze(text=text, language="en")

        return self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "DEFAULT": OperatorConfig("encrypt", {"key": key})
            }
        ).text

# الاستخدام
anonymizer = PIIAnonymizer()

original = "Email me at jane@corp.com, my SSN is 123-45-6789"
masked = anonymizer.mask(original)
# المخرجات: "Email me at [بريد], my SSN is [محذوف]"

التكامل غير المتزامن للإنتاج

Presidio ليس غير متزامن أصلياً، لكن يمكننا تغليفه بكفاءة:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict

class AsyncPresidio:
    """غلاف غير متزامن لـ Presidio مع تجميع الاتصالات."""

    def __init__(self, max_workers: int = 4):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    async def analyze(self, text: str, entities: List[str] = None) -> List[Dict]:
        """تحليل كيانات غير متزامن."""
        loop = asyncio.get_event_loop()
        results = await loop.run_in_executor(
            self.executor,
            lambda: self.analyzer.analyze(
                text=text,
                entities=entities,
                language="en"
            )
        )
        return [
            {
                "entity_type": r.entity_type,
                "start": r.start,
                "end": r.end,
                "score": r.score,
                "text": text[r.start:r.end]
            }
            for r in results
        ]

    async def anonymize(self, text: str, strategy: str = "mask") -> str:
        """إخفاء هوية غير متزامن."""
        loop = asyncio.get_event_loop()

        def _anonymize():
            results = self.analyzer.analyze(text=text, language="en")
            return self.anonymizer.anonymize(
                text=text,
                analyzer_results=results
            ).text

        return await loop.run_in_executor(self.executor, _anonymize)

# الاستخدام في خط أنابيب الحواجز
async def pii_filter_layer(user_input: str) -> tuple[str, Dict]:
    presidio = AsyncPresidio()

    # اكتشاف الكيانات
    entities = await presidio.analyze(user_input)

    if not entities:
        return user_input, {"pii_detected": False}

    # إخفاء الهوية
    cleaned = await presidio.anonymize(user_input)

    return cleaned, {
        "pii_detected": True,
        "entities_found": len(entities),
        "entity_types": list(set(e["entity_type"] for e in entities))
    }

المتعرفات المخصصة لـ PII الخاص بالمجال

from presidio_analyzer import PatternRecognizer, Pattern

# متعرف مخصص لمعرفات الموظفين الداخلية
employee_id_recognizer = PatternRecognizer(
    supported_entity="EMPLOYEE_ID",
    patterns=[
        Pattern(
            name="employee_id",
            regex=r"EMP-\d{6}",
            score=0.9
        )
    ]
)

# إضافة للسجل
registry = RecognizerRegistry()
registry.load_predefined_recognizers()
registry.add_recognizer(employee_id_recognizer)

# إنشاء محلل مع سجل مخصص
custom_analyzer = AnalyzerEngine(registry=registry)

# الآن يكتشف الكيانات المخصصة
text = "Contact EMP-123456 for assistance"
results = custom_analyzer.analyze(text=text, language="en")
# يكتشف: EMPLOYEE_ID: EMP-123456

نصيحة إنتاجية: Presidio عادة يعالج النص في 20-50 مللي ثانية، مما يجعله سريعاً بما يكفي لطبقة التصفية قبل LLM. للمعالجة الدفعية، استخدم نمط ThreadPoolExecutor الموضح أعلاه.

التالي: بناء خطوط أنابيب اكتشاف حقن المحثات. :::