Coverage for src / local_deep_research / embeddings / providers / base.py: 94%

1"""Base class for embedding providers.""" 

2 

3from abc import ABC, abstractmethod 

4from typing import Any, Dict, List, Optional 

5 

6from langchain_core.embeddings import Embeddings 

7 

8 

9class BaseEmbeddingProvider(ABC): 

10 """ 

11 Abstract base class for embedding providers. 

12 

13 All embedding providers should inherit from this class and implement 

14 the required methods. This provides a consistent interface similar to 

15 the LLM provider system. 

16 """ 

17 

18 # Override these in subclasses 

19 provider_name = "base" # Display name for logs/UI 

20 provider_key = "BASE" # Unique identifier (uppercase) 

21 requires_api_key = False # Whether this provider requires an API key 

22 supports_local = False # Whether this runs locally 

23 default_model = None # Default embedding model 

24 

25 @classmethod 

26 @abstractmethod 

27 def create_embeddings( 

28 cls, 

29 model: Optional[str] = None, 

30 settings_snapshot: Optional[Dict[str, Any]] = None, 

31 **kwargs, 

32 ) -> Embeddings: 

33 """ 

34 Create an embeddings instance for this provider. 

35 

36 Args: 

37 model: Name of the embedding model to use 

38 settings_snapshot: Optional settings snapshot for thread-safe access 

39 **kwargs: Additional provider-specific parameters 

40 

41 Returns: 

42 A LangChain Embeddings instance 

43 

44 Raises: 

45 ValueError: If required configuration is missing 

46 ImportError: If required dependencies are not installed 

47 """ 

48 pass 

49 

50 @classmethod 

51 @abstractmethod 

52 def is_available( 

53 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

54 ) -> bool: 

55 """ 

56 Check if this embedding provider is available and properly configured. 

57 

58 Args: 

59 settings_snapshot: Optional settings snapshot for thread-safe access 

60 

61 Returns: 

62 True if the provider can be used, False otherwise 

63 """ 

64 pass 

65 

66 @classmethod 

67 def get_available_models( 

68 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

69 ) -> List[Dict[str, str]]: 

70 """ 

71 Get list of available models for this provider. 

72 

73 Args: 

74 settings_snapshot: Optional settings snapshot 

75 

76 Returns: 

77 List of dicts with 'value' and 'label' keys for each model 

78 """ 

79 return [] 

80 

81 @classmethod 

82 def get_model_info(cls, model: str) -> Optional[Dict[str, Any]]: 

83 """ 

84 Get information about a specific model. 

85 

86 Args: 

87 model: Model identifier 

88 

89 Returns: 

90 Dict with model metadata (dimensions, description, etc.) or None 

91 """ 

92 return None 

93 

94 @classmethod 

95 def validate_config( 

96 cls, settings_snapshot: Optional[Dict[str, Any]] = None 

97 ) -> tuple[bool, Optional[str]]: 

98 """ 

99 Validate the provider configuration. 

100 

101 Args: 

102 settings_snapshot: Optional settings snapshot 

103 

104 Returns: 

105 Tuple of (is_valid, error_message) 

106 """ 

107 if not cls.is_available(settings_snapshot): 

108 return ( 

109 False, 

110 f"{cls.provider_name} is not available or not configured", 

111 ) 

112 return True, None 

113 

114 @classmethod 

115 def get_provider_info(cls) -> Dict[str, Any]: 

116 """ 

117 Get metadata about this provider. 

118 

119 Returns: 

120 Dict with provider information 

121 """ 

122 return { 

123 "name": cls.provider_name, 

124 "key": cls.provider_key, 

125 "requires_api_key": cls.requires_api_key, 

126 "supports_local": cls.supports_local, 

127 "default_model": cls.default_model, 

128 }