Coverage for presidio_analyzer / predefined_recognizers / third_party / basic_langextract_recognizer.py: 96%

1import logging 

2import os 

3from pathlib import Path 

4from typing import Any, Dict, Optional 

5 

6from presidio_analyzer.llm_utils import lx_factory 

7from presidio_analyzer.predefined_recognizers.third_party.\ 

8 langextract_recognizer import LangExtractRecognizer 

9 

10logger = logging.getLogger("presidio-analyzer") 

11 

12DEFAULT_EXTRACT_PARAMS = { 

13 "max_char_buffer": 400, 

14 "use_schema_constraints": False, 

15 "fence_output": False, 

16} 

17 

18DEFAULT_LANGUAGE_MODEL_PARAMS = { 

19 "timeout": 240, 

20 "num_ctx": 8192 

21} 

22 

23class BasicLangExtractRecognizer(LangExtractRecognizer): 

24 """Basic LangExtract recognizer using configurable backend.""" 

25 

26 DEFAULT_CONFIG_PATH = ( 

27 Path(__file__).parent.parent.parent / "conf" / "langextract_config_basic.yaml" 

28 ) 

29 

30 def __init__( 

31 self, 

32 config_path: Optional[str] = None, 

33 supported_language: str = "en", 

34 context: Optional[list] = None, 

35 name="BasicLangExtractRecognizer", 

36 **kwargs 

37 ): 

38 """Initialize Basic LangExtract recognizer. 

39 

40 :param config_path: Path to configuration file (optional). 

41 :param supported_language: Language this recognizer supports 

42 (optional, default: "en"). 

43 :param context: List of context words 

44 (optional, currently not used by LLM recognizers). 

45 """ 

46 actual_config_path = ( 

47 config_path if config_path else str(self.DEFAULT_CONFIG_PATH) 

48 ) 

49 

50 super().__init__( 

51 config_path=actual_config_path, 

52 name=name, 

53 supported_language=supported_language, 

54 extract_params={ 

55 "extract": DEFAULT_EXTRACT_PARAMS, 

56 "language_model": DEFAULT_LANGUAGE_MODEL_PARAMS 

57 } 

58 ) 

59 

60 model_config: Dict[str, Any] = self.config.get("model", {}) 

61 provider_config = model_config.get("provider", {}) 

62 

63 self.model_id = model_config.get("model_id") 

64 self.provider = provider_config.get("name") 

65 self.provider_kwargs = provider_config.get("kwargs", {}) 

66 

67 # Not ideal, but update _extract_params now that self.config is fully loaded. 

68 self._extract_params.update(provider_config.get("extract_params", {})) 

69 self._language_model_params.update( 

70 provider_config.get("language_model_params", {}) 

71 ) 

72 

73 if not self.provider: 

74 raise ValueError("Configuration must contain " 

75 "'langextract.model.provider.name'") 

76 

77 if ("api_key" not in self.provider_kwargs 

78 and "LANGEXTRACT_API_KEY" in os.environ): 

79 self.provider_kwargs["api_key"] = os.environ["LANGEXTRACT_API_KEY"] 

80 

81 self.lx_model_config = lx_factory.ModelConfig( 

82 model_id=self.model_id, 

83 provider=self.provider, 

84 provider_kwargs=self.provider_kwargs, 

85 ) 

86 

87 def _get_provider_params(self): 

88 """Return supplementary params.""" 

89 return { 

90 "config": self.lx_model_config, 

91 }