agentic_workflow/pii.py

from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine

_provider = NlpEngineProvider(nlp_configuration={
    "nlp_engine_name": "spacy",
    "models": [{"lang_code": "en", "model_name": "en_core_web_lg"}]
})

analyzer  = AnalyzerEngine(nlp_engine=_provider.create_engine())
anonymizer = AnonymizerEngine()

analyzer.registry.add_recognizer(PatternRecognizer(
    supported_entity="ACCOUNT_NUMBER",
    patterns=[Pattern(name="account_number", regex=r"\d{3}-\d{3}-\d{3}", score=0.8)],
    context=["account", "acct", "num", "number", "no.", "#"]
))

_ENTITIES = ["ACCOUNT_NUMBER", "PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "CREDIT_CARD"]

# 2 funcs
def check_pii(text: str) -> bool:
    """Returns True if PII is detected — used by planner to block the query."""
    results = analyzer.analyze(text=text, language="en", entities=_ENTITIES)
    return len(results) > 0

def scrub_pii(text: str) -> str:
    """Replaces PII with placeholders — use this if you want to proceed with a sanitised query instead of blocking."""
    results = analyzer.analyze(text=text, language="en", entities=_ENTITIES)
    return anonymizer.anonymize(text=text, analyzer_results=results).text