diff --git a/core/src/agentforge_core/guardrails/validators.py b/core/src/agentforge_core/guardrails/validators.py index b0d128b..26b0b3a 100644 --- a/core/src/agentforge_core/guardrails/validators.py +++ b/core/src/agentforge_core/guardrails/validators.py @@ -21,6 +21,33 @@ _INJECTION_PATTERNS = [ r"reveal (the )?(system|hidden) (prompt|instruction)", ] +# Singleton perezoso del AnalyzerEngine de Presidio. Construirlo es caro (carga el +# modelo spaCy y los recognizers), así que se reutiliza entre llamadas. +_PRESIDIO_ANALYZER: Any = None + + +def _presidio_analyzer() -> Any: + """``AnalyzerEngine`` de Presidio configurado con el modelo spaCy ``en_core_web_sm``. + + Presidio usa por defecto ``en_core_web_lg`` (~560 MB), que **no** está en la imagen + Docker: ``core/Dockerfile`` instala ``en_core_web_sm``. Si Presidio no está instalado + o el modelo no se puede cargar, esto lanza y el llamador (`detect_pii`) hace fallback + a la detección por regex. + """ + global _PRESIDIO_ANALYZER + if _PRESIDIO_ANALYZER is None: + from presidio_analyzer import AnalyzerEngine + from presidio_analyzer.nlp_engine import NlpEngineProvider + + nlp_engine = NlpEngineProvider( + nlp_configuration={ + "nlp_engine_name": "spacy", + "models": [{"lang_code": "en", "model_name": "en_core_web_sm"}], + } + ).create_engine() + _PRESIDIO_ANALYZER = AnalyzerEngine(nlp_engine=nlp_engine) + return _PRESIDIO_ANALYZER + def _violation( *, @@ -45,11 +72,11 @@ def _violation( def detect_pii( text: str, config: dict[str, Any], trace_id: UUID, stage: str ) -> list[GuardrailViolation]: - """Detección PII vía Presidio Analyzer (con fallback a regex si no está instalado).""" + """Detección PII vía Presidio Analyzer (con fallback a regex si no está disponible).""" try: - from presidio_analyzer import AnalyzerEngine + analyzer = _presidio_analyzer() except Exception: - # Si Presidio no está, fallback a regex básica + # Presidio no instalado o modelo spaCy no disponible → fallback a regex básica. return _pii_regex_fallback(text, config, trace_id, stage) entities = config.get( @@ -59,7 +86,6 @@ def detect_pii( sev = config.get("severity_on_match", "block") blocked = sev == "block" - analyzer = AnalyzerEngine() results = analyzer.analyze(text=text, entities=entities, language="en") if not results: return []